use std::{
    collections::{BTreeSet, HashSet},
    fs,
};

use iroh::SecretKey;
use tempfile::TempDir;
use uuid::Uuid;

use super::{FolderSession, MAX_AUDIT_EVENTS_PER_RESPONSE, NodeHost, protect_live_blobs};

fn test_roster(
    folder_id: Uuid,
    active_member_ids: impl IntoIterator<Item = String>,
) -> crate::domain::FolderRoster {
    let owner = SecretKey::generate();
    let member = crate::domain::RosterMember {
        device_id: owner.public().to_string(),
        display_name: None,
        role: crate::domain::MemberRole::Owner,
    };
    let mut roster =
        crate::domain::FolderRoster::create(folder_id, "capability".to_owned(), member)
            .expect("roster");
    for device_id in active_member_ids {
        roster = roster
            .add_member(crate::domain::RosterMember {
                device_id,
                display_name: None,
                role: crate::domain::MemberRole::Member,
            })
            .expect("add member");
    }
    roster.sign(&owner).expect("sign roster")
}

#[tokio::test]
async fn transfers_a_file_between_shared_blob_stores() -> anyhow::Result<()> {
    let source_directory = TempDir::new()?;
    let target_directory = TempDir::new()?;
    let source_file = source_directory.path().join("hello.txt");
    let target_file = target_directory.path().join("received.txt");
    fs::write(&source_file, "hello from Appa")?;
    let source =
        NodeHost::load_with_lan_discovery(source_directory.path(), SecretKey::generate(), false)
            .await?;
    let target =
        NodeHost::load_with_lan_discovery(target_directory.path(), SecretKey::generate(), false)
            .await?;
    let (blob_hash, _temp_tag) = source.import_file(&source_file).await?;
    source.wait_until_online().await?;
    target.wait_until_online().await?;
    target
        .download_blob(blob_hash, source.endpoint_address())
        .await?;
    target.export_blob(blob_hash, &target_file).await?;
    assert_eq!(fs::read_to_string(target_file)?, "hello from Appa");
    source.shutdown().await?;
    target.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn transfers_tiny_files_in_one_blob_request() -> anyhow::Result<()> {
    let source_directory = TempDir::new()?;
    let target_directory = TempDir::new()?;
    let source =
        NodeHost::load_with_lan_discovery(source_directory.path(), SecretKey::generate(), false)
            .await?;
    let target =
        NodeHost::load_with_lan_discovery(target_directory.path(), SecretKey::generate(), false)
            .await?;
    let first_source = source_directory.path().join("first.txt");
    let second_source = source_directory.path().join("second.txt");
    fs::write(&first_source, "first")?;
    fs::write(&second_source, "second")?;
    let (first_hash, _first_tag) = source.import_file(&first_source).await?;
    let (second_hash, _second_tag) = source.import_file(&second_source).await?;
    source.wait_until_online().await?;
    target.wait_until_online().await?;

    target
        .download_blobs(vec![first_hash, second_hash], source.endpoint_address())
        .await?;
    target
        .export_blob(first_hash, &target_directory.path().join("first.txt"))
        .await?;
    target
        .export_blob(second_hash, &target_directory.path().join("second.txt"))
        .await?;

    assert_eq!(
        fs::read_to_string(target_directory.path().join("first.txt"))?,
        "first"
    );
    assert_eq!(
        fs::read_to_string(target_directory.path().join("second.txt"))?,
        "second"
    );
    source.shutdown().await?;
    target.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn import_file_uses_temp_tags_not_persistent_named_tags() -> anyhow::Result<()> {
    use futures_util::StreamExt;

    let directory = TempDir::new()?;
    let node =
        NodeHost::load_with_lan_discovery(directory.path(), SecretKey::generate(), false).await?;
    let file = directory.path().join("hello.txt");
    fs::write(&file, "protected content")?;

    let (hash, _temp_tag) = node.import_file(&file).await?;
    assert!(node.store.blobs().has(hash).await?);

    // import_file must not leave a persistent named tag behind; only temp tags
    // should protect the blob so GC can reclaim it once the manifest references it.
    let named_tags: Vec<_> = node.store.tags().list().await?.collect().await;
    assert!(
        named_tags.is_empty(),
        "import_file should not create persistent named tags"
    );

    node.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn sweep_legacy_tags_removes_persistent_named_tags() -> anyhow::Result<()> {
    use futures_util::StreamExt;

    let directory = TempDir::new()?;
    let node =
        NodeHost::load_with_lan_discovery(directory.path(), SecretKey::generate(), false).await?;
    let file = directory.path().join("legacy.txt");
    fs::write(&file, "old import style")?;

    // Simulate the old behavior: add_path().await creates a persistent named tag.
    let tag_info = node.store.blobs().add_path(&file).await?;
    let named_before: Vec<_> = node.store.tags().list().await?.collect().await;
    assert_eq!(named_before.len(), 1);

    let swept = node.sweep_legacy_tags().await?;
    assert_eq!(swept, 1);
    let named_after: Vec<_> = node.store.tags().list().await?.collect().await;
    assert!(named_after.is_empty());

    drop(tag_info);
    node.shutdown().await?;
    Ok(())
}

#[test]
fn gc_protection_aborts_when_the_state_database_cannot_be_read() {
    let directory = TempDir::new().expect("temporary directory");
    let mut live = HashSet::new();

    assert!(matches!(
        protect_live_blobs(directory.path(), &mut live),
        iroh_blobs::store::ProtectOutcome::Abort
    ));
    assert!(live.is_empty());
}

#[tokio::test]
async fn announces_folder_changes_to_an_authorized_peer() -> anyhow::Result<()> {
    let source_directory = TempDir::new()?;
    let target_directory = TempDir::new()?;
    let source =
        NodeHost::load_with_lan_discovery(source_directory.path(), SecretKey::generate(), false)
            .await?;
    let target =
        NodeHost::load_with_lan_discovery(target_directory.path(), SecretKey::generate(), false)
            .await?;
    let folder_id = Uuid::new_v4();
    let capability = "folder-capability".to_owned();
    target
        .register_folder(
            FolderSession {
                folder_id,
                capability: capability.clone(),
                active_member_ids: BTreeSet::from([source.endpoint_address().id.to_string()]),
                roster: test_roster(folder_id, [source.endpoint_address().id.to_string()]),
                peer_endpoints: vec![source.endpoint_address()],
                audit_events: vec![],
            },
            crate::domain::Manifest::empty(folder_id),
        )
        .await?;
    source.wait_until_online().await?;
    target.wait_until_online().await?;

    source
        .announce_manifest(
            vec![target.endpoint_address()],
            folder_id,
            capability,
            "root-one".to_owned(),
        )
        .await;
    source
        .announce_manifest(
            vec![target.endpoint_address()],
            folder_id,
            "folder-capability".to_owned(),
            "root-two".to_owned(),
        )
        .await;

    assert_eq!(target.take_announced_folders(), vec![folder_id]);
    assert!(target.take_announced_folders().is_empty());
    source.shutdown().await?;
    target.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn requests_remote_merkle_children_missing_from_a_small_local_manifest() -> anyhow::Result<()>
{
    let source_directory = TempDir::new()?;
    let target_directory = TempDir::new()?;
    let source =
        NodeHost::load_with_lan_discovery(source_directory.path(), SecretKey::generate(), false)
            .await?;
    let target =
        NodeHost::load_with_lan_discovery(target_directory.path(), SecretKey::generate(), false)
            .await?;
    let folder_id = Uuid::new_v4();
    let capability = "folder-capability".to_owned();
    let remote_manifest = large_manifest(folder_id);
    source
        .register_folder(
            FolderSession {
                folder_id,
                capability: capability.clone(),
                active_member_ids: BTreeSet::from([target.endpoint_address().id.to_string()]),
                roster: test_roster(folder_id, [target.endpoint_address().id.to_string()]),
                peer_endpoints: vec![target.endpoint_address()],
                audit_events: vec![],
            },
            remote_manifest.clone(),
        )
        .await?;
    source.wait_until_online().await?;
    target.wait_until_online().await?;

    let received = target
        .request_manifest_delta(
            source.endpoint_address(),
            folder_id,
            capability,
            None,
            &crate::domain::Manifest::empty(folder_id),
        )
        .await?;

    assert_eq!(received, remote_manifest);
    source.shutdown().await?;
    target.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn retrieves_a_complete_audit_log_in_pages() -> anyhow::Result<()> {
    let source_directory = TempDir::new()?;
    let target_directory = TempDir::new()?;
    let source =
        NodeHost::load_with_lan_discovery(source_directory.path(), SecretKey::generate(), false)
            .await?;
    let target =
        NodeHost::load_with_lan_discovery(target_directory.path(), SecretKey::generate(), false)
            .await?;
    let folder_id = Uuid::new_v4();
    let capability = "folder-capability".to_owned();
    let audit_identity = SecretKey::generate();
    let roster = signed_owner_roster(folder_id, &audit_identity, &capability)?;
    let audit_events = signed_audit_events(
        folder_id,
        &audit_identity,
        &roster,
        MAX_AUDIT_EVENTS_PER_RESPONSE + 1,
    )?;
    source
        .register_folder(
            FolderSession {
                folder_id,
                capability: capability.clone(),
                active_member_ids: BTreeSet::from([target.endpoint_address().id.to_string()]),
                roster,
                peer_endpoints: vec![target.endpoint_address()],
                audit_events,
            },
            crate::domain::Manifest::empty(folder_id),
        )
        .await?;
    source.wait_until_online().await?;
    target.wait_until_online().await?;

    let first_page = target
        .request_audit_events(
            source.endpoint_address(),
            folder_id,
            capability.clone(),
            None,
            Default::default(),
        )
        .await?;
    let last_sequence = first_page.last().expect("first page has events").sequence;
    let second_page = target
        .request_audit_events(
            source.endpoint_address(),
            folder_id,
            capability,
            None,
            std::collections::BTreeMap::from([(
                audit_identity.public().to_string(),
                last_sequence,
            )]),
        )
        .await?;

    assert_eq!(first_page.len(), MAX_AUDIT_EVENTS_PER_RESPONSE);
    assert_eq!(second_page.len(), 1);
    assert_eq!(second_page[0].sequence, last_sequence + 1);
    source.shutdown().await?;
    target.shutdown().await?;
    Ok(())
}

fn signed_owner_roster(
    folder_id: Uuid,
    identity: &SecretKey,
    capability: &str,
) -> anyhow::Result<crate::domain::FolderRoster> {
    crate::domain::FolderRoster::create(
        folder_id,
        capability.to_owned(),
        crate::domain::RosterMember {
            device_id: identity.public().to_string(),
            display_name: None,
            role: crate::domain::MemberRole::Owner,
        },
    )?
    .sign(identity)
}

fn signed_audit_events(
    folder_id: Uuid,
    identity: &SecretKey,
    roster: &crate::domain::FolderRoster,
    count: usize,
) -> anyhow::Result<Vec<crate::domain::AuditEvent>> {
    let mut events = Vec::with_capacity(count);
    let mut parent_hash = None;
    for sequence in 1..=u64::try_from(count)? {
        let event = crate::domain::AuditEvent::create(
            folder_id,
            identity.public().to_string(),
            sequence,
            parent_hash,
            crate::domain::AuditEventKind::ManifestCommitted,
            Some(format!("manifest-{sequence}")),
            roster.hash()?,
            roster.clone(),
            time::OffsetDateTime::UNIX_EPOCH,
        )
        .sign(identity)?;
        parent_hash = Some(event.hash()?);
        events.push(event);
    }
    Ok(events)
}

fn large_manifest(folder_id: Uuid) -> crate::domain::Manifest {
    let mut manifest = crate::domain::Manifest::empty(folder_id);
    for number in 0..65 {
        let path = format!("file-{number}.txt");
        manifest.entries.insert(
            path.clone(),
            crate::domain::Entry {
                path,
                kind: crate::domain::EntryKind::File,
                blob_hash: Some(format!("blob-{number}")),
                size_bytes: Some(1),
                modified_at: time::OffsetDateTime::UNIX_EPOCH,
                clock: Default::default(),
                author_device_id: "source".to_owned(),
            },
        );
    }
    manifest
}

#[tokio::test]
async fn serves_multiple_isolated_folders_from_one_endpoint() -> anyhow::Result<()> {
    let source_directory = TempDir::new()?;
    let target_directory = TempDir::new()?;
    let source =
        NodeHost::load_with_lan_discovery(source_directory.path(), SecretKey::generate(), false)
            .await?;
    let target =
        NodeHost::load_with_lan_discovery(target_directory.path(), SecretKey::generate(), false)
            .await?;
    let third_directory = TempDir::new()?;
    let third =
        NodeHost::load_with_lan_discovery(third_directory.path(), SecretKey::generate(), false)
            .await?;
    let first_folder = Uuid::new_v4();
    let second_folder = Uuid::new_v4();
    source
        .register_folder(
            FolderSession {
                folder_id: first_folder,
                capability: "first".to_owned(),
                active_member_ids: BTreeSet::from([
                    target.endpoint_address().id.to_string(),
                    third.endpoint_address().id.to_string(),
                ]),
                roster: test_roster(
                    first_folder,
                    [
                        target.endpoint_address().id.to_string(),
                        third.endpoint_address().id.to_string(),
                    ],
                ),
                peer_endpoints: vec![target.endpoint_address(), third.endpoint_address()],
                audit_events: vec![],
            },
            crate::domain::Manifest::empty(first_folder),
        )
        .await?;
    source
        .register_folder(
            FolderSession {
                folder_id: second_folder,
                capability: "second".to_owned(),
                active_member_ids: BTreeSet::from([target.endpoint_address().id.to_string()]),
                roster: test_roster(second_folder, [target.endpoint_address().id.to_string()]),
                peer_endpoints: vec![],
                audit_events: vec![],
            },
            crate::domain::Manifest::empty(second_folder),
        )
        .await?;
    source.wait_until_online().await?;
    target.wait_until_online().await?;
    third.wait_until_online().await?;
    let first_response = target
        .request_manifest_summary(
            source.endpoint_address(),
            first_folder,
            "first".to_owned(),
            None,
        )
        .await?;
    assert_eq!(first_response.summary.folder_id, first_folder);
    assert_eq!(first_response.peer_endpoints.len(), 2);
    assert_eq!(
        first_response.peer_endpoints[0].id,
        target.endpoint_address().id
    );
    let third_response = third
        .request_manifest_summary(
            source.endpoint_address(),
            first_folder,
            "first".to_owned(),
            None,
        )
        .await?;
    assert!(
        third_response
            .peer_endpoints
            .iter()
            .any(|endpoint| endpoint.id == target.endpoint_address().id)
    );
    assert_eq!(
        target
            .request_manifest_summary(
                source.endpoint_address(),
                second_folder,
                "second".to_owned(),
                None,
            )
            .await?
            .summary
            .folder_id,
        second_folder
    );
    assert!(
        target
            .request_manifest_summary(
                source.endpoint_address(),
                first_folder,
                "second".to_owned(),
                None,
            )
            .await
            .is_err()
    );
    let rotated_folder = Uuid::new_v4();
    source
        .register_folder(
            FolderSession {
                folder_id: rotated_folder,
                capability: "new-capability".to_owned(),
                active_member_ids: BTreeSet::from([target.endpoint_address().id.to_string()]),
                roster: test_roster(rotated_folder, [target.endpoint_address().id.to_string()]),
                peer_endpoints: vec![],
                audit_events: vec![],
            },
            crate::domain::Manifest::empty(rotated_folder),
        )
        .await?;
    assert!(
        target
            .request_manifest_summary(
                source.endpoint_address(),
                rotated_folder,
                "old-capability".to_owned(),
                None,
            )
            .await
            .is_err()
    );
    source.shutdown().await?;
    target.shutdown().await?;
    third.shutdown().await?;
    Ok(())
}