use std::{
    collections::{BTreeMap, BTreeSet, VecDeque},
    fs,
    path::Path,
    str::FromStr,
};

use anyhow::Context;
use futures_util::{StreamExt, stream};

use crate::{
    app::{MAX_CONCURRENT_BLOB_TRANSFERS, SyncContext, manifest::deleted_entry},
    domain::{DeviceId, Entry, EntryKind, Manifest, ReconciliationAction},
    filesystem::{IgnoreRules, remove_path, safe_destination},
    iroh::NodeHost,
    storage::FolderConfig,
};

pub(super) struct RemoteManifestResult {
    pub(super) manifest: Manifest,
    pub(super) synchronized_file_count: usize,
}

pub(super) async fn apply_remote_manifest(
    context: &SyncContext<'_>,
    local: &Manifest,
    remote: &Manifest,
    peer: iroh::EndpointAddr,
) -> anyhow::Result<RemoteManifestResult> {
    let mut merged = local.clone();
    let mut synchronized_count = 0;
    let ignore_rules = IgnoreRules::load(&context.folder.path)?;
    let actions = crate::domain::reconcile_manifests(local, remote);
    let (mut materialization_plan, downloads) =
        plan_remote_materialization(&ignore_rules, context.folder, actions, remote);
    materialize_ready_actions(
        context,
        &mut merged,
        &mut synchronized_count,
        &mut materialization_plan,
    )
    .await?;
    receive_and_materialize_files(
        context,
        &peer,
        &mut merged,
        &mut synchronized_count,
        &mut materialization_plan,
        downloads,
    )
    .await?;
    Ok(RemoteManifestResult {
        manifest: merged,
        synchronized_file_count: synchronized_count,
    })
}

pub(super) async fn restore_manifest(
    context: &SyncContext<'_>,
    current: &Manifest,
    historical: &Manifest,
) -> anyhow::Result<usize> {
    let paths = current
        .entries
        .keys()
        .chain(historical.entries.keys())
        .collect::<std::collections::BTreeSet<_>>();
    let device_id = context.node.endpoint_address().id.to_string();
    let mut restored = Manifest::empty(context.folder.id);
    let mut file_writes = 0;
    for path in paths {
        let current_entry = current.entries.get(path);
        let historical_entry = historical.entries.get(path);
        let Some(entry) =
            restored_entry(path, current_entry, historical_entry, &device_id, context)?
        else {
            continue;
        };
        if current_entry != Some(&entry) {
            file_writes += materialize_entries(context, &mut restored, vec![entry]).await?;
        } else {
            restored.entries.insert(path.clone(), entry);
        }
    }
    context.state_store.save_manifest(&restored)?;
    Ok(file_writes)
}

fn restored_entry(
    path: &str,
    current: Option<&Entry>,
    historical: Option<&Entry>,
    device_id: &DeviceId,
    context: &SyncContext<'_>,
) -> anyhow::Result<Option<Entry>> {
    let Some(historical) = historical else {
        return current
            .cloned()
            .map(|entry| deleted_entry(path.to_owned(), entry, device_id, context))
            .transpose();
    };
    let mut restored = historical.clone();
    let mut clock = current
        .map(|entry| entry.clock.clone())
        .unwrap_or_else(|| historical.clock.clone());
    clock.insert(
        device_id.clone(),
        context
            .state_store
            .next_counter(context.folder.id, device_id)?,
    );
    restored.clock = clock;
    restored.author_device_id = device_id.clone();
    Ok(Some(restored))
}

fn action_is_ignored(
    folder: &FolderConfig,
    ignore_rules: &IgnoreRules,
    action: &ReconciliationAction,
) -> bool {
    let entry = match action {
        ReconciliationAction::ApplyRemote { entry } | ReconciliationAction::KeepLocal { entry } => {
            entry
        }
        ReconciliationAction::ResolveConflict { winner, .. } => winner,
    };
    ignore_rules.ignores(
        &folder.path.join(&entry.path),
        entry.kind == EntryKind::Directory,
    )
}

pub(super) async fn recover_pending_materialization(
    context: &SyncContext<'_>,
) -> anyhow::Result<()> {
    let Some(pending) = context
        .state_store
        .pending_materialization(context.folder.id)?
    else {
        return Ok(());
    };
    if pending.is_legacy_single_entry {
        tracing::warn!("Discarding a stale single-entry materialization journal");
        context
            .state_store
            .clear_pending_materialization(context.folder.id)?;
        return Ok(());
    }
    if pending.resulting_manifest.folder_id != context.folder.id {
        anyhow::bail!("pending materialization belongs to another folder");
    }
    tracing::warn!(
        entry_count = pending.entries.len(),
        "Recovering interrupted filesystem updates"
    );
    for entry in &pending.entries {
        apply_entry(context, entry).await?;
    }
    context
        .state_store
        .save_manifest(&pending.resulting_manifest)?;
    context
        .state_store
        .clear_pending_materialization(context.folder.id)?;
    Ok(())
}

async fn materialize_entries(
    context: &SyncContext<'_>,
    manifest: &mut Manifest,
    entries: Vec<Entry>,
) -> anyhow::Result<usize> {
    let mut resulting_manifest = manifest.clone();
    for entry in &entries {
        resulting_manifest
            .entries
            .insert(entry.path.clone(), entry.clone());
    }
    context.state_store.save_pending_materialization(
        context.folder.id,
        &entries,
        &resulting_manifest,
    )?;
    let mut writes = 0;
    for entry in &entries {
        writes += apply_entry(context, entry).await?;
    }
    context.state_store.save_manifest(&resulting_manifest)?;
    context
        .state_store
        .clear_pending_materialization(context.folder.id)?;
    *manifest = resulting_manifest;
    Ok(writes)
}

async fn apply_entry(context: &SyncContext<'_>, entry: &Entry) -> anyhow::Result<usize> {
    let destination = safe_destination(&context.folder.path, &entry.path)?;
    match entry.kind {
        EntryKind::Deleted => {
            remove_path(&destination)?;
            Ok(0)
        }
        EntryKind::Directory => {
            replace_non_directory(&destination)?;
            fs::create_dir_all(destination)?;
            Ok(0)
        }
        EntryKind::File => {
            let blob_hash = iroh_blobs::Hash::from_str(
                entry
                    .blob_hash
                    .as_deref()
                    .ok_or_else(|| anyhow::anyhow!("file entry has no blob hash"))?,
            )?;
            remove_existing_destination(&destination)?;
            write_blob_atomically(context.node, blob_hash, &destination).await?;
            tracing::info!(path = %entry.path, "Wrote file");
            Ok(1)
        }
    }
}

fn replace_non_directory(destination: &Path) -> anyhow::Result<()> {
    let metadata = match fs::symlink_metadata(destination) {
        Ok(metadata) => metadata,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
        Err(error) => return Err(error.into()),
    };
    if metadata.is_dir() {
        return Ok(());
    }
    remove_path(destination)
}

fn remove_existing_destination(destination: &Path) -> anyhow::Result<()> {
    match fs::symlink_metadata(destination) {
        Ok(_) => remove_path(destination),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(error.into()),
    }
}

struct RemoteFile {
    path: String,
    blob_hash: String,
    size_bytes: Option<u64>,
}

const SMALL_BLOB_MAX_BYTES: u64 = 256 * 1024;
const MAX_SMALL_BLOBS_PER_REQUEST: usize = 128;
const MAX_SMALL_BLOB_REQUEST_BYTES: u64 = 4 * 1024 * 1024;

struct PendingAction {
    action: ReconciliationAction,
    missing_blob_count: usize,
}

struct MaterializationPlan {
    pending_actions: Vec<Option<PendingAction>>,
    actions_waiting_by_blob: BTreeMap<String, Vec<usize>>,
    ready_actions: VecDeque<usize>,
}

fn plan_remote_materialization(
    ignore_rules: &IgnoreRules,
    folder: &FolderConfig,
    actions: Vec<ReconciliationAction>,
    remote: &Manifest,
) -> (MaterializationPlan, Vec<RemoteFile>) {
    let mut files = BTreeMap::new();
    let mut pending_actions = Vec::new();
    let mut actions_waiting_by_blob = BTreeMap::new();
    let mut ready_actions = VecDeque::new();
    for action in actions
        .into_iter()
        .filter(|action| !action_is_ignored(folder, ignore_rules, action))
    {
        let required_blobs = collect_required_blobs(&mut files, remote, &action);
        let action_id = pending_actions.len();
        if required_blobs.is_empty() {
            ready_actions.push_back(action_id);
        } else {
            for blob_hash in &required_blobs {
                actions_waiting_by_blob
                    .entry(blob_hash.clone())
                    .or_insert_with(Vec::new)
                    .push(action_id);
            }
        }
        pending_actions.push(Some(PendingAction {
            action,
            missing_blob_count: required_blobs.len(),
        }));
    }
    (
        MaterializationPlan {
            pending_actions,
            actions_waiting_by_blob,
            ready_actions,
        },
        files.into_values().collect(),
    )
}

fn collect_required_blobs(
    files: &mut BTreeMap<String, RemoteFile>,
    remote: &Manifest,
    action: &ReconciliationAction,
) -> BTreeSet<String> {
    let mut required_blobs = BTreeSet::new();
    for entry in remote_entries_for_action(action) {
        collect_remote_file(files, &mut required_blobs, remote, entry);
    }
    required_blobs
}

fn remote_entries_for_action(action: &ReconciliationAction) -> Vec<&Entry> {
    match action {
        ReconciliationAction::ApplyRemote { entry } => vec![entry],
        ReconciliationAction::KeepLocal { .. } => Vec::new(),
        ReconciliationAction::ResolveConflict { winner, loser, .. } => vec![winner, loser],
    }
}

async fn materialize_ready_actions(
    context: &SyncContext<'_>,
    merged: &mut Manifest,
    synchronized_count: &mut usize,
    plan: &mut MaterializationPlan,
) -> anyhow::Result<()> {
    while let Some(actions) = plan.take_ready_actions() {
        *synchronized_count += materialize_action_batch(context, merged, actions).await?;
    }
    Ok(())
}

impl MaterializationPlan {
    fn note_completed_blob(&mut self, blob_hash: String) {
        let Some(waiting_actions) = self.actions_waiting_by_blob.remove(&blob_hash) else {
            return;
        };
        for action_id in waiting_actions {
            let Some(pending) = self.pending_actions[action_id].as_mut() else {
                continue;
            };
            pending.missing_blob_count -= 1;
            if pending.missing_blob_count == 0 {
                self.ready_actions.push_back(action_id);
            }
        }
    }

    fn take_ready_actions(&mut self) -> Option<Vec<ReconciliationAction>> {
        const MATERIALIZATION_BATCH_SIZE: usize = 64;
        let mut actions = Vec::new();
        while actions.len() < MATERIALIZATION_BATCH_SIZE {
            let Some(action_id) = self.ready_actions.pop_front() else {
                break;
            };
            let Some(pending) = self.pending_actions[action_id].take() else {
                continue;
            };
            actions.push(pending.action);
        }
        (!actions.is_empty()).then_some(actions)
    }
}

async fn materialize_action_batch(
    context: &SyncContext<'_>,
    merged: &mut Manifest,
    actions: Vec<ReconciliationAction>,
) -> anyhow::Result<usize> {
    let entries = actions
        .into_iter()
        .flat_map(entries_for_action)
        .collect::<Vec<_>>();
    if entries.is_empty() {
        return Ok(0);
    }
    materialize_entries(context, merged, entries).await
}

fn entries_for_action(action: ReconciliationAction) -> Vec<Entry> {
    match action {
        ReconciliationAction::ApplyRemote { entry } => vec![entry],
        ReconciliationAction::KeepLocal { .. } => Vec::new(),
        ReconciliationAction::ResolveConflict {
            winner,
            loser,
            conflict_path,
        } => {
            let mut conflict_entry = loser;
            conflict_entry.path = conflict_path;
            vec![conflict_entry, winner]
        }
    }
}

async fn receive_and_materialize_files(
    context: &SyncContext<'_>,
    peer: &iroh::EndpointAddr,
    merged: &mut Manifest,
    synchronized_count: &mut usize,
    plan: &mut MaterializationPlan,
    downloads: Vec<RemoteFile>,
) -> anyhow::Result<()> {
    // Protect every blob we are about to download from garbage collection until
    // the manifest references it (which happens inside `materialize_entries`).
    let _download_guards = protect_download_hashes(context.node, &downloads).await;
    let mut downloads = stream::iter(plan_download_batches(downloads))
        .map(|files| download_files_with_isolation(context, peer, files))
        .buffer_unordered(MAX_CONCURRENT_BLOB_TRANSFERS);
    while let Some(result) = downloads.next().await {
        match result {
            Ok(blob_hashes) => {
                for blob_hash in blob_hashes {
                    plan.note_completed_blob(blob_hash);
                }
                materialize_ready_actions(context, merged, synchronized_count, plan).await?;
            }
            Err(error) => {
                tracing::warn!(error = ?error, "Could not receive file; deferring dependent changes")
            }
        }
    }
    Ok(())
}

async fn protect_download_hashes(
    node: &NodeHost,
    downloads: &[RemoteFile],
) -> Vec<crate::iroh::TempTag> {
    let hashes = downloads
        .iter()
        .filter_map(|file| iroh_blobs::Hash::from_str(&file.blob_hash).ok())
        .collect::<Vec<_>>();
    match node.protect_blobs(&hashes).await {
        Ok(guards) => guards,
        Err(error) => {
            tracing::warn!(%error, "Could not protect download blobs from GC; relying on GC timing");
            Vec::new()
        }
    }
}

fn plan_download_batches(downloads: Vec<RemoteFile>) -> Vec<Vec<RemoteFile>> {
    let mut batches = Vec::new();
    let mut small_files = Vec::new();
    let mut small_file_bytes = 0;
    for file in downloads {
        if !is_small_file(&file) {
            push_small_file_batch(&mut batches, &mut small_files, &mut small_file_bytes);
            batches.push(vec![file]);
            continue;
        }
        if small_files.len() == MAX_SMALL_BLOBS_PER_REQUEST
            || small_file_bytes + file.size_bytes.unwrap_or_default() > MAX_SMALL_BLOB_REQUEST_BYTES
        {
            push_small_file_batch(&mut batches, &mut small_files, &mut small_file_bytes);
        }
        small_file_bytes += file.size_bytes.unwrap_or_default();
        small_files.push(file);
    }
    push_small_file_batch(&mut batches, &mut small_files, &mut small_file_bytes);
    batches
}

fn is_small_file(file: &RemoteFile) -> bool {
    file.size_bytes
        .is_some_and(|size| size <= SMALL_BLOB_MAX_BYTES)
}

fn push_small_file_batch(
    batches: &mut Vec<Vec<RemoteFile>>,
    small_files: &mut Vec<RemoteFile>,
    small_file_bytes: &mut u64,
) {
    if small_files.is_empty() {
        return;
    }
    batches.push(std::mem::take(small_files));
    *small_file_bytes = 0;
}

async fn download_files_with_isolation(
    context: &SyncContext<'_>,
    peer: &iroh::EndpointAddr,
    files: Vec<RemoteFile>,
) -> anyhow::Result<Vec<String>> {
    let mut completed_hashes = Vec::new();
    let mut pending_batches = VecDeque::from([files]);
    while let Some(batch) = pending_batches.pop_front() {
        match download_file_batch(context, peer, &batch).await {
            Ok(hashes) => completed_hashes.extend(hashes),
            Err(error) if batch.len() == 1 => {
                tracing::warn!(path = %batch[0].path, error = ?error, "Could not receive file; deferring dependent changes");
            }
            Err(error) => {
                tracing::debug!(file_count = batch.len(), error = ?error, "Splitting failed blob request");
                // A failed multi-file request does not identify the bad blob.
                // Halving preserves transfers that are independent of it.
                let midpoint = batch.len() / 2;
                let mut first_half = batch;
                let second_half = first_half.split_off(midpoint);
                pending_batches.push_back(first_half);
                pending_batches.push_back(second_half);
            }
        }
    }
    Ok(completed_hashes)
}

async fn download_file_batch(
    context: &SyncContext<'_>,
    peer: &iroh::EndpointAddr,
    files: &[RemoteFile],
) -> anyhow::Result<Vec<String>> {
    for file in files {
        tracing::debug!(path = %file.path, size_bytes = ?file.size_bytes, "Receiving file");
    }
    let hashes = files
        .iter()
        .map(|file| iroh_blobs::Hash::from_str(&file.blob_hash).map_err(anyhow::Error::from))
        .collect::<anyhow::Result<Vec<_>>>()?;
    if let [blob_hash] = hashes.as_slice() {
        context
            .node
            .download_blob(*blob_hash, peer.clone())
            .await
            .with_context(|| format!("could not receive {}", files[0].path))?;
    } else {
        context
            .node
            .download_blobs(hashes, peer.clone())
            .await
            .with_context(|| format!("could not receive {} files", files.len()))?;
    }
    Ok(files.iter().map(|file| file.blob_hash.clone()).collect())
}

fn collect_remote_file(
    files: &mut BTreeMap<String, RemoteFile>,
    required_blobs: &mut BTreeSet<String>,
    remote: &Manifest,
    entry: &Entry,
) {
    if remote.entries.get(&entry.path) != Some(entry) || entry.kind != EntryKind::File {
        return;
    }
    let Some(blob_hash) = entry.blob_hash.clone() else {
        return;
    };
    required_blobs.insert(blob_hash.clone());
    files.entry(blob_hash.clone()).or_insert(RemoteFile {
        path: entry.path.clone(),
        blob_hash,
        size_bytes: entry.size_bytes,
    });
}

async fn write_blob_atomically(
    node: &NodeHost,
    blob_hash: iroh_blobs::Hash,
    destination: &Path,
) -> anyhow::Result<()> {
    let parent = destination
        .parent()
        .ok_or_else(|| anyhow::anyhow!("destination has no parent directory"))?;
    fs::create_dir_all(parent)?;
    let temporary_file = tempfile::NamedTempFile::new_in(parent)?;
    node.export_blob(blob_hash, temporary_file.path()).await?;
    let (_, temporary_path) = temporary_file.keep()?;
    fs::rename(temporary_path, destination)?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use std::fs;

    use tempfile::TempDir;
    use time::OffsetDateTime;
    use uuid::Uuid;

    use crate::{
        domain::{Entry, EntryKind, FolderMode, Manifest, ReconciliationAction},
        filesystem::IgnoreRules,
        storage::FolderConfig,
    };

    use super::{
        MAX_SMALL_BLOBS_PER_REQUEST, RemoteFile, plan_download_batches,
        plan_remote_materialization, remove_existing_destination, replace_non_directory,
    };

    #[test]
    fn replaces_a_file_with_a_directory() -> anyhow::Result<()> {
        let directory = TempDir::new()?;
        let destination = directory.path().join("item");
        fs::write(&destination, "file")?;

        replace_non_directory(&destination)?;
        fs::create_dir(&destination)?;

        assert!(destination.is_dir());
        Ok(())
    }

    #[test]
    fn removes_a_directory_before_writing_a_file() -> anyhow::Result<()> {
        let directory = TempDir::new()?;
        let destination = directory.path().join("item");
        fs::create_dir(&destination)?;
        fs::write(destination.join("child"), "contents")?;

        remove_existing_destination(&destination)?;

        assert!(!destination.exists());
        Ok(())
    }

    #[test]
    fn plans_independent_file_actions_separately() -> anyhow::Result<()> {
        let directory = TempDir::new()?;
        let folder_id = Uuid::new_v4();
        let first = file_entry("first.txt", "first-hash");
        let second = file_entry("second.txt", "second-hash");
        let remote = Manifest {
            folder_id,
            entries: [
                (first.path.clone(), first.clone()),
                (second.path.clone(), second.clone()),
            ]
            .into(),
        };
        let actions = vec![
            ReconciliationAction::ApplyRemote { entry: first },
            ReconciliationAction::ApplyRemote { entry: second },
        ];
        let folder = FolderConfig {
            id: folder_id,
            name: "test".to_owned(),
            path: directory.path().to_owned(),
            capability: "test-capability".to_owned(),
            mode: FolderMode::SendReceive,
        };

        let (materialization_plan, downloads) = plan_remote_materialization(
            &IgnoreRules::load(directory.path())?,
            &folder,
            actions,
            &remote,
        );

        assert_eq!(materialization_plan.pending_actions.len(), 2);
        assert_eq!(
            materialization_plan.actions_waiting_by_blob["first-hash"],
            vec![0]
        );
        assert_eq!(
            materialization_plan.actions_waiting_by_blob["second-hash"],
            vec![1]
        );
        assert_eq!(downloads.len(), 2);
        Ok(())
    }

    #[test]
    fn batches_tiny_files_without_mixing_in_large_files() {
        let tiny_files = (0..=MAX_SMALL_BLOBS_PER_REQUEST)
            .map(|number| RemoteFile {
                path: format!("tiny-{number}"),
                blob_hash: format!("hash-{number}"),
                size_bytes: Some(1),
            })
            .collect();
        let downloads = plan_download_batches(tiny_files);

        assert_eq!(downloads.len(), 2);
        assert_eq!(downloads[0].len(), MAX_SMALL_BLOBS_PER_REQUEST);
        assert_eq!(downloads[1].len(), 1);

        let downloads = plan_download_batches(vec![
            RemoteFile {
                path: "tiny".to_owned(),
                blob_hash: "tiny-hash".to_owned(),
                size_bytes: Some(1),
            },
            RemoteFile {
                path: "large".to_owned(),
                blob_hash: "large-hash".to_owned(),
                size_bytes: Some(257 * 1024),
            },
        ]);

        assert_eq!(downloads.iter().map(Vec::len).collect::<Vec<_>>(), [1, 1]);
    }

    fn file_entry(path: &str, blob_hash: &str) -> Entry {
        Entry {
            path: path.to_owned(),
            kind: EntryKind::File,
            blob_hash: Some(blob_hash.to_owned()),
            size_bytes: Some(1),
            modified_at: OffsetDateTime::UNIX_EPOCH,
            clock: Default::default(),
            author_device_id: "device".to_owned(),
        }
    }
}