vstorage 0.7.0

// Copyright 2023-2025 Hugo Osvaldo Barrera
//
// SPDX-License-Identifier: EUPL-1.2

mod common;

use futures_util::stream::TryStreamExt;
use std::path::Path;
use std::sync::Arc;
use tempfile::tempdir;
use vstorage::ItemKind;
use vstorage::base::{CreateItemOptions, Storage};
use vstorage::sync::plan::Plan;
use vstorage::sync::{
    conflict::{KeepSideResolver, resolve_conflicts},
    declare::{OnDelete, OnEmpty, StoragePair, SyncedCollection},
    execute::Executor,
    operation::{ItemOp, Operation},
    status::{Side, StatusDatabase},
};
use vstorage::vdir::VdirStorage;

use common::{minimal_icalendar, minimal_icalendar_with_uid};

/// Create a storage with three calendars each one with a single event.
async fn create_populated_storage(path: &Path) -> Arc<dyn Storage> {
    let storage = VdirStorage::builder(path.to_path_buf().try_into().unwrap())
        .unwrap()
        .build(ItemKind::Calendar);

    let first = storage.create_collection("first-calendar").await.unwrap();
    let item = &minimal_icalendar("First calendar event one")
        .unwrap()
        .into();
    let opts = CreateItemOptions::default();
    storage
        .create_item(first.href(), item, opts.clone())
        .await
        .unwrap();

    let item = &minimal_icalendar("First calendar event two")
        .unwrap()
        .into();
    storage
        .create_item(first.href(), item, opts.clone())
        .await
        .unwrap();
    drop(first);

    let second = storage.create_collection("second-calendar").await.unwrap();
    let item = &minimal_icalendar("Second calendar event one")
        .unwrap()
        .into();
    storage
        .create_item(second.href(), item, opts.clone())
        .await
        .unwrap();

    let item = &minimal_icalendar("Second calendar event two")
        .unwrap()
        .into();
    storage
        .create_item(second.href(), item, opts.clone())
        .await
        .unwrap();
    drop(second);

    let third = storage.create_collection("third-calendar").await.unwrap();
    let item = &minimal_icalendar("Third calendar event one")
        .unwrap()
        .into();
    storage.create_item(third.href(), item, opts).await.unwrap();
    drop(third);

    Arc::new(storage)
}

fn create_empty_storage(path: &Path) -> Arc<dyn Storage> {
    let storage = VdirStorage::builder(path.to_path_buf().try_into().unwrap())
        .unwrap()
        .build(ItemKind::Calendar);
    Arc::new(storage)
}

// TODO: maybe "second calendar" should exist in side B for these to make more sense.

#[tokio::test]
async fn sync_only_declared_mappings() {
    let populated_path = tempdir().unwrap();
    let empty_path = tempdir().unwrap();
    let populated = create_populated_storage(populated_path.path()).await;
    let empty = create_empty_storage(empty_path.path());

    let first_mapping = SyncedCollection::direct("first-calendar".parse().unwrap());
    let second_mapping = SyncedCollection::direct("second-calendar".parse().unwrap());
    // third-calendar is not synced.

    let pair = StoragePair::new(populated.clone(), empty.clone())
        .with_mapping(first_mapping)
        .with_mapping(second_mapping);

    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();

    Executor::new(drop)
        .execute_stream(populated, empty, operations, &status)
        .await
        .unwrap()
        .unwrap();

    let first = std::fs::read_dir(empty_path.path().join("first-calendar"))
        .unwrap()
        .map(|r| r.unwrap())
        .collect::<Vec<_>>();
    assert_eq!(first.len(), 2);
    for item in first {
        let data = std::fs::read_to_string(item.path()).unwrap();
        let found = data.find("First calendar event");
        assert!(found.is_some());
    }

    let second = std::fs::read_dir(empty_path.path().join("second-calendar"))
        .unwrap()
        .map(|r| r.unwrap())
        .collect::<Vec<_>>();
    assert_eq!(second.len(), 2);
    for item in second {
        let data = std::fs::read_to_string(item.path()).unwrap();
        let found = data.find("Second calendar event");
        assert!(found.is_some());
    }

    let _third = std::fs::read_dir(empty_path.path().join("third-calendar")).unwrap_err();

    std::fs::remove_dir_all(populated_path).unwrap();
    std::fs::remove_dir_all(empty_path).unwrap();
}

#[tokio::test]
async fn sync_from_a() {
    let populated_path = tempdir().unwrap();
    let empty_path = tempdir().unwrap();
    let populated = create_populated_storage(populated_path.path()).await;
    let empty = create_empty_storage(empty_path.path());

    let pair = StoragePair::new(populated.clone(), empty.clone()).with_all_from_a();
    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();

    Executor::new(drop)
        .execute_stream(populated, empty, operations, &status)
        .await
        .unwrap()
        .unwrap();

    let first = std::fs::read_dir(empty_path.path().join("first-calendar"))
        .unwrap()
        .map(|r| r.unwrap())
        .collect::<Vec<_>>();

    // Note that "third-calendar" is not present in the "empty" one.
    assert_eq!(first.len(), 2);

    for item in first {
        let data = std::fs::read_to_string(item.path()).unwrap();
        let found = data.find("First calendar event");
        assert!(found.is_some());
    }

    let second = std::fs::read_dir(empty_path.path().join("second-calendar"))
        .unwrap()
        .map(|r| r.unwrap())
        .collect::<Vec<_>>();
    assert_eq!(second.len(), 2);
    for item in second {
        let data = std::fs::read_to_string(item.path()).unwrap();
        let found = data.find("Second calendar event");
        assert!(found.is_some());
    }

    let third = std::fs::read_dir(empty_path.path().join("third-calendar"))
        .unwrap()
        .map(|r| r.unwrap())
        .collect::<Vec<_>>();
    assert_eq!(third.len(), 1);
    for item in third {
        let data = std::fs::read_to_string(item.path()).unwrap();
        let found = data.find("Third calendar event");
        assert!(found.is_some());
    }

    std::fs::remove_dir_all(populated_path).unwrap();
    std::fs::remove_dir_all(empty_path).unwrap();
}

#[tokio::test]
async fn sync_from_b() {
    let populated_path = tempdir().unwrap();
    let empty_path = tempdir().unwrap();
    let populated = create_populated_storage(populated_path.path()).await;
    let empty = create_empty_storage(empty_path.path());

    let pair = StoragePair::new(populated.clone(), empty.clone()).with_all_from_b();
    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();

    Executor::new(drop)
        .execute_stream(populated, empty, operations, &status)
        .await
        .unwrap()
        .unwrap();

    let _first = std::fs::read_dir(empty_path.path().join("first-calendar")).unwrap_err();
    let _second = std::fs::read_dir(empty_path.path().join("second-calendar")).unwrap_err();
    let _third = std::fs::read_dir(empty_path.path().join("third-calendar")).unwrap_err();

    std::fs::remove_dir_all(populated_path).unwrap();
    std::fs::remove_dir_all(empty_path).unwrap();
}

#[tokio::test]
async fn sync_none() {
    let populated_path = tempdir().unwrap();
    let empty_path = tempdir().unwrap();
    let populated = create_populated_storage(populated_path.path()).await;
    let empty = create_empty_storage(empty_path.path());

    let pair = StoragePair::new(populated.clone(), empty.clone());
    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();

    Executor::new(drop)
        .execute_stream(populated, empty, operations, &status)
        .await
        .unwrap()
        .unwrap();

    let _first = std::fs::read_dir(empty_path.path().join("first-calendar")).unwrap_err();
    let _second = std::fs::read_dir(empty_path.path().join("second-calendar")).unwrap_err();
    let _third = std::fs::read_dir(empty_path.path().join("third-calendar")).unwrap_err();

    std::fs::remove_dir_all(populated_path).unwrap();
    std::fs::remove_dir_all(empty_path).unwrap();
}

#[tokio::test]
async fn sync_deletion_from_a() {
    let path_a = tempdir().unwrap();
    let path_b = tempdir().unwrap();
    let storage_a = Arc::new(
        VdirStorage::builder(path_a.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );
    let storage_b = Arc::new(
        VdirStorage::builder(path_b.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );

    storage_a.create_collection("my-calendar").await.unwrap();
    storage_b.create_collection("my-calendar").await.unwrap();
    let item = &minimal_icalendar("First calendar event one")
        .unwrap()
        .into();
    let opts = CreateItemOptions {
        resource_name: Some("hello.ics".to_string()),
    };
    let item_a = storage_a
        .create_item("my-calendar", item, opts.clone())
        .await
        .unwrap();
    storage_b
        .create_item("my-calendar", item, opts)
        .await
        .unwrap();

    let pair = StoragePair::new(storage_a.clone(), storage_b.clone())
        .with_all_from_b()
        .on_empty(OnEmpty::Sync);
    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let operations = Plan::new(pair.clone(), Some(status.clone())).await.unwrap();

    Executor::new(drop)
        .execute_stream(storage_a.clone(), storage_b.clone(), operations, &status)
        .await
        .unwrap()
        .unwrap();

    // Both storages in sync with an updated status database.

    storage_a
        .delete_item(&item_a.href, &item_a.etag)
        .await
        .unwrap();

    let operations = Plan::new(pair.clone(), Some(status.clone())).await.unwrap();

    let ops: Vec<Operation> = operations.try_collect().await.unwrap();
    let has_delete_b = ops
        .iter()
        .any(|op| matches!(op, Operation::Item(ItemOp::Delete(d)) if d.side == Side::B));
    assert!(has_delete_b, "Should generate Delete(side: B) operation");

    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();
    Executor::new(drop)
        .execute_stream(storage_a.clone(), storage_b.clone(), operations, &status)
        .await
        .unwrap()
        .unwrap();

    let count = storage_b.get_all_items("my-calendar").await.unwrap().len();
    assert_eq!(count, 0);
}

#[tokio::test]
async fn sync_deletion_from_b() {
    let path_a = tempdir().unwrap();
    let path_b = tempdir().unwrap();
    let storage_a = Arc::new(
        VdirStorage::builder(path_a.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );
    let storage_b = Arc::new(
        VdirStorage::builder(path_b.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );

    storage_a.create_collection("my-calendar").await.unwrap();
    storage_b.create_collection("my-calendar").await.unwrap();
    let item = &minimal_icalendar("First calendar event one")
        .unwrap()
        .into();
    let opts = CreateItemOptions {
        resource_name: Some("hello.ics".to_string()),
    };
    storage_a
        .create_item("my-calendar", item, opts.clone())
        .await
        .unwrap();
    let item_b = storage_b
        .create_item("my-calendar", item, opts)
        .await
        .unwrap();

    let pair = StoragePair::new(storage_a.clone(), storage_b.clone())
        .with_all_from_b()
        .on_empty(OnEmpty::Sync);
    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let operations = Plan::new(pair.clone(), Some(status.clone())).await.unwrap();

    Executor::new(drop)
        .execute_stream(storage_a.clone(), storage_b.clone(), operations, &status)
        .await
        .unwrap()
        .unwrap();

    // Both storages in sync with an updated status database.

    storage_b
        .delete_item(&item_b.href, &item_b.etag)
        .await
        .unwrap();

    let operations = Plan::new(pair.clone(), Some(status.clone())).await.unwrap();

    let ops: Vec<Operation> = operations.try_collect().await.unwrap();
    let has_delete_a = ops
        .iter()
        .any(|op| matches!(op, Operation::Item(ItemOp::Delete(d)) if d.side == Side::A));
    assert!(has_delete_a, "Should generate Delete(side: A) operation");

    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();
    Executor::new(drop)
        .execute_stream(storage_a.clone(), storage_b.clone(), operations, &status)
        .await
        .unwrap()
        .unwrap();

    let count = storage_a.get_all_items("my-calendar").await.unwrap().len();
    assert_eq!(count, 0);
}

#[tokio::test]
async fn sync_creation_from_a() {
    let path_a = tempdir().unwrap();
    let path_b = tempdir().unwrap();
    let storage_a = Arc::new(
        VdirStorage::builder(path_a.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );
    let storage_b = Arc::new(
        VdirStorage::builder(path_b.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );

    storage_a.create_collection("my-calendar").await.unwrap();
    let item = &minimal_icalendar("First calendar event one")
        .unwrap()
        .into();
    storage_a
        .create_item("my-calendar", item, CreateItemOptions::default())
        .await
        .unwrap();

    let pair = StoragePair::new(storage_a.clone(), storage_b.clone()).with_all_from_a();
    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();

    let ops: Vec<Operation> = operations.try_collect().await.unwrap();
    let has_write_in_b = ops
        .iter()
        .any(|op| matches!(op, Operation::Item(ItemOp::Write(w)) if w.target_side == Side::B));
    assert!(
        has_write_in_b,
        "Should generate Write(target_side: B) operation"
    );

    let pair = StoragePair::new(storage_a.clone(), storage_b.clone()).with_all_from_a();
    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();
    Executor::new(drop)
        .execute_stream(storage_a, storage_b.clone(), operations, &status)
        .await
        .unwrap()
        .unwrap();

    let fetched = &storage_b.get_all_items("my-calendar").await.unwrap()[0];
    assert_eq!(fetched.item.as_str(), item.as_str());
}

#[tokio::test]
async fn sync_creation_from_b() {
    let path_a = tempdir().unwrap();
    let path_b = tempdir().unwrap();
    let storage_a = Arc::new(
        VdirStorage::builder(path_a.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );
    let storage_b = Arc::new(
        VdirStorage::builder(path_b.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );

    storage_b.create_collection("my-calendar").await.unwrap();
    let item = &minimal_icalendar("First calendar event one")
        .unwrap()
        .into();
    storage_b
        .create_item("my-calendar", item, CreateItemOptions::default())
        .await
        .unwrap();

    let pair = StoragePair::new(storage_a.clone(), storage_b.clone()).with_all_from_b();
    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();

    let ops: Vec<Operation> = operations.try_collect().await.unwrap();
    let has_write_in_a = ops
        .iter()
        .any(|op| matches!(op, Operation::Item(ItemOp::Write(w)) if w.target_side == Side::A));
    assert!(
        has_write_in_a,
        "Should generate Write(target_side: A) operation"
    );

    let pair = StoragePair::new(storage_a.clone(), storage_b.clone()).with_all_from_b();
    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();
    Executor::new(drop)
        .execute_stream(storage_a.clone(), storage_b, operations, &status)
        .await
        .unwrap()
        .unwrap();

    let fetched = &storage_a.get_all_items("my-calendar").await.unwrap()[0];
    assert_eq!(fetched.item.as_str(), item.as_str());
}

#[tokio::test]
async fn empty_on_empty_skip() {
    let path_a = tempdir().unwrap();
    let path_b = tempdir().unwrap();
    let storage_a = Arc::new(
        VdirStorage::builder(path_a.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );
    let storage_b = Arc::new(
        VdirStorage::builder(path_b.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );

    let first = storage_a.create_collection("first-calendar").await.unwrap();
    let item = &minimal_icalendar("First calendar event one")
        .unwrap()
        .into();
    let opts = CreateItemOptions::default();
    let item_ver = storage_a
        .create_item(first.href(), item, opts)
        .await
        .unwrap();

    let pair = StoragePair::new(storage_a.clone(), storage_b.clone())
        .with_all_from_a()
        .on_empty(OnEmpty::Skip);
    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let operations = Plan::new(pair.clone(), Some(status.clone())).await.unwrap();

    Executor::new(drop)
        .execute_stream(storage_a.clone(), storage_b.clone(), operations, &status)
        .await
        .unwrap()
        .unwrap();

    // At this point both storages and the status DB are all in sync.

    storage_a
        .delete_item(&item_ver.href, &item_ver.etag)
        .await
        .unwrap();

    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();

    let ops: Vec<Operation> = operations.try_collect().await.unwrap();
    assert!(
        ops.is_empty(),
        "Should generate no operations due to OnEmpty::Skip"
    );
}

#[tokio::test]
async fn empty_on_empty_sync() {
    let path_a = tempdir().unwrap();
    let path_b = tempdir().unwrap();
    let storage_a = Arc::new(
        VdirStorage::builder(path_a.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );
    let storage_b = Arc::new(
        VdirStorage::builder(path_b.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );

    let first = storage_a.create_collection("first-calendar").await.unwrap();
    let item = &minimal_icalendar("First calendar event one")
        .unwrap()
        .into();
    let opts = CreateItemOptions::default();
    let item_ver = storage_a
        .create_item(first.href(), item, opts)
        .await
        .unwrap();

    let pair = StoragePair::new(storage_a.clone(), storage_b.clone())
        .with_all_from_a()
        .on_empty(OnEmpty::Sync);
    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let operations = Plan::new(pair.clone(), Some(status.clone())).await.unwrap();

    Executor::new(drop)
        .execute_stream(storage_a.clone(), storage_b.clone(), operations, &status)
        .await
        .unwrap()
        .unwrap();

    // At this point both storages and the status DB are all in sync.

    storage_a
        .delete_item(&item_ver.href, &item_ver.etag)
        .await
        .unwrap();

    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();

    let ops: Vec<Operation> = operations.try_collect().await.unwrap();
    let has_delete_b = ops
        .iter()
        .any(|op| matches!(op, Operation::Item(ItemOp::Delete(d)) if d.side == Side::B));
    assert!(has_delete_b, "Should generate Delete(side: B) operation");
}

#[tokio::test]
async fn empty_on_delete_skip() {
    let path_a = tempdir().unwrap();
    let path_b = tempdir().unwrap();
    let storage_a = Arc::new(
        VdirStorage::builder(path_a.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );
    let storage_b = Arc::new(
        VdirStorage::builder(path_b.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );

    let first = storage_a.create_collection("first-calendar").await.unwrap();

    let pair = StoragePair::new(storage_a.clone(), storage_b.clone())
        .with_all_from_a()
        .with_all_from_b()
        .on_delete(OnDelete::Skip);
    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let operations = Plan::new(pair.clone(), Some(status.clone())).await.unwrap();

    Executor::new(drop)
        .execute_stream(storage_a.clone(), storage_b.clone(), operations, &status)
        .await
        .unwrap()
        .unwrap();

    // At this point both storages and the status DB are all in sync.

    storage_a.delete_collection(first.href()).await.unwrap();

    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();

    let ops: Vec<Operation> = operations.try_collect().await.unwrap();
    assert!(
        ops.is_empty(),
        "Should generate no operations due to OnDelete::Skip"
    );
}

#[tokio::test]
async fn empty_on_delete_sync() {
    let path_a = tempdir().unwrap();
    let path_b = tempdir().unwrap();
    let storage_a = Arc::new(
        VdirStorage::builder(path_a.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );
    let storage_b = Arc::new(
        VdirStorage::builder(path_b.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );

    let first = storage_a.create_collection("first-calendar").await.unwrap();

    let pair = StoragePair::new(storage_a.clone(), storage_b.clone())
        .with_all_from_a()
        .with_all_from_b()
        .on_delete(OnDelete::Sync);
    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let operations = Plan::new(pair.clone(), Some(status.clone())).await.unwrap();

    Executor::new(drop)
        .execute_stream(storage_a.clone(), storage_b.clone(), operations, &status)
        .await
        .unwrap()
        .unwrap();

    // At this point both storages and the status DB are all in sync.

    storage_a.delete_collection(first.href()).await.unwrap();

    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();

    let ops: Vec<Operation> = operations.try_collect().await.unwrap();
    let has_collection_delete = ops.iter().any(|op| {
        matches!(op, Operation::Collection(vstorage::sync::operation::CollectionOp::Delete { side, .. }) if *side == Side::B)
    });
    assert!(
        has_collection_delete,
        "Should generate collection Delete operation for side B"
    );
}

/// Test conflict resolution when there's no prior sync (old: None).
#[tokio::test]
async fn conflict_resolution_first_sync() {
    let path_a = tempdir().unwrap();
    let path_b = tempdir().unwrap();

    // Create storage A with an item.
    let storage_a = VdirStorage::builder(path_a.path().to_path_buf().try_into().unwrap())
        .unwrap()
        .build(ItemKind::Calendar);
    let storage_a = Arc::new(storage_a);

    let collection_a = storage_a.create_collection("calendar").await.unwrap();
    let item_a = minimal_icalendar_with_uid("conflict-test-uid", "Event from A")
        .unwrap()
        .into();
    storage_a
        .create_item(collection_a.href(), &item_a, CreateItemOptions::default())
        .await
        .unwrap();

    // Create storage B with an item. Same UID but conflicting content.
    let storage_b = VdirStorage::builder(path_b.path().to_path_buf().try_into().unwrap())
        .unwrap()
        .build(ItemKind::Calendar);
    let storage_b = Arc::new(storage_b);

    let collection_b = storage_b.create_collection("calendar").await.unwrap();
    let item_b = minimal_icalendar_with_uid("conflict-test-uid", "Event from B")
        .unwrap()
        .into();
    storage_b
        .create_item(collection_b.href(), &item_b, CreateItemOptions::default())
        .await
        .unwrap();

    // Setup sync with conflict resolution

    let pair = StoragePair::new(storage_a.clone(), storage_b.clone())
        .with_mapping(SyncedCollection::direct("calendar".parse().unwrap()));

    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();
    let resolved = resolve_conflicts(operations, KeepSideResolver(Side::A));

    // Execute the sync
    Executor::new(drop)
        .execute_stream(storage_a.clone(), storage_b.clone(), resolved, &status)
        .await
        .unwrap()
        .unwrap();

    // Side B should now have content from side A.

    let items_b = storage_b.get_all_items(collection_b.href()).await.unwrap();
    assert_eq!(items_b.len(), 1);
    let content_b = items_b[0].item.as_str();
    assert!(content_b.contains("Event from A"),);
    assert!(!content_b.contains("Event from B"),);

    // Side A should still have its original content.

    let items_a = storage_a.get_all_items(collection_a.href()).await.unwrap();
    assert_eq!(items_a.len(), 1);
    let content_a = items_a[0].item.as_str();
    assert!(content_a.contains("Event from A"),);
}

/// Test that sync properly iterates iterating through all items.
#[tokio::test]
async fn sync_continues_after_unchanged_items() {
    let dir_a = tempdir().unwrap();
    let dir_b = tempdir().unwrap();

    let storage_a = Arc::new(
        VdirStorage::builder(dir_a.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );
    let storage_b = Arc::new(
        VdirStorage::builder(dir_b.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );

    let collection_a = storage_a.create_collection("calendar").await.unwrap();
    let collection_b = storage_b.create_collection("calendar").await.unwrap();

    // Create the same three items on both sides.
    let uid1 = "item-one-uid";
    let uid2 = "item-two-uid";
    let uid3 = "item-three-uid";

    let opts = CreateItemOptions::default();

    let item1_a = minimal_icalendar_with_uid(uid1, "Event One")
        .unwrap()
        .into();
    let item2_a = minimal_icalendar_with_uid(uid2, "Event Two")
        .unwrap()
        .into();
    let item3_a = minimal_icalendar_with_uid(uid3, "Event Three")
        .unwrap()
        .into();

    storage_a
        .create_item(collection_a.href(), &item1_a, opts.clone())
        .await
        .unwrap();
    storage_a
        .create_item(collection_a.href(), &item2_a, opts.clone())
        .await
        .unwrap();
    storage_a
        .create_item(collection_a.href(), &item3_a, opts.clone())
        .await
        .unwrap();

    storage_b
        .create_item(collection_b.href(), &item1_a, opts.clone())
        .await
        .unwrap();
    storage_b
        .create_item(collection_b.href(), &item2_a, opts.clone())
        .await
        .unwrap();
    storage_b
        .create_item(collection_b.href(), &item3_a, opts.clone())
        .await
        .unwrap();

    let pair = StoragePair::new(storage_a.clone(), storage_b.clone())
        .with_mapping(SyncedCollection::direct("calendar".parse().unwrap()));

    // Establish baseline in status database.
    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let plan = Plan::new(pair.clone(), Some(status.clone())).await.unwrap();

    Executor::new(drop)
        .execute_stream(storage_a.clone(), storage_b.clone(), plan, &status)
        .await
        .unwrap()
        .unwrap();

    // Modify item3 on side A.
    let items_a = storage_a.get_all_items(collection_a.href()).await.unwrap();
    let item3_ver = items_a
        .iter()
        .find(|i| i.item.ident() == uid3)
        .expect("item3 should exist");

    let item3_modified = minimal_icalendar_with_uid(uid3, "Event Three MODIFIED")
        .unwrap()
        .into();
    storage_a
        .update_item(&item3_ver.href, &item3_ver.etag, &item3_modified)
        .await
        .unwrap();

    // Items 1 and 2 are unchanged (no operation), item 3 is modified
    let plan = Plan::new(pair, Some(status.clone())).await.unwrap();

    // Collect all operations to verify the correct ones are generated.
    let operations: Vec<_> = plan.try_collect().await.unwrap();

    // Should have exactly one item operation: Write(target_side: B) for item3.
    assert_eq!(operations.len(), 1);
    match &operations[0] {
        Operation::Item(ItemOp::Write(write)) => {
            assert_eq!(write.target_side, Side::B);
            assert!(
                write
                    .source
                    .data
                    .as_ref()
                    .expect("Write should have data")
                    .as_str()
                    .contains("Event Three MODIFIED"),
                "Write should contain the modified content"
            );
        }
        other => panic!("Expected Write operation, got {:?}", other),
    }

    let plan = Plan::new(
        StoragePair::new(storage_a.clone(), storage_b.clone())
            .with_mapping(SyncedCollection::direct("calendar".parse().unwrap())),
        Some(status.clone()),
    )
    .await
    .unwrap();

    Executor::new(drop)
        .execute_stream(storage_a.clone(), storage_b.clone(), plan, &status)
        .await
        .unwrap()
        .unwrap();

    // Verify item3 is updated on side B.
    let items_b = storage_b.get_all_items(collection_b.href()).await.unwrap();
    let item3_b = items_b
        .iter()
        .find(|i| i.item.ident() == uid3)
        .expect("item3 should exist on side B");

    assert!(item3_b.item.as_str().contains("Event Three MODIFIED"),);
}

/// Syncing an item from A to B retains the original filename.
#[tokio::test]
async fn sync_retains_filename() {
    let path_a = tempdir().unwrap();
    let path_b = tempdir().unwrap();
    let storage_a = Arc::new(
        VdirStorage::builder(path_a.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );
    let storage_b = Arc::new(
        VdirStorage::builder(path_b.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );

    storage_a.create_collection("my-calendar").await.unwrap();

    // Item filename ("custom-name.ics") differs from its UID ("my-uid-123").
    let item = &minimal_icalendar_with_uid("my-uid-123", "Test Event")
        .unwrap()
        .into();
    let opts = CreateItemOptions {
        resource_name: Some("custom-name".to_string()),
    };
    storage_a
        .create_item("my-calendar", item, opts)
        .await
        .unwrap();

    let pair = StoragePair::new(storage_a.clone(), storage_b.clone()).with_all_from_a();
    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();

    Executor::new(|err| panic!("Sync error: {err:?}"))
        .execute_stream(storage_a, storage_b.clone(), operations, &status)
        .await
        .unwrap()
        .unwrap();

    let items_b = storage_b.get_all_items("my-calendar").await.unwrap();
    assert_eq!(items_b.len(), 1);
    assert!(
        items_b[0].href.ends_with("custom-name.ics"),
        "Expected href ending with 'custom-name.ics', got: {}",
        items_b[0].href
    );
}

/// Syncing items with the same filename but different UIDs falls back to a
/// default name for the colliding item, so both items end up on both sides.
#[tokio::test]
async fn sync_with_filename_collision() {
    let path_a = tempdir().unwrap();
    let path_b = tempdir().unwrap();
    let storage_a = Arc::new(
        VdirStorage::builder(path_a.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );
    let storage_b = Arc::new(
        VdirStorage::builder(path_b.path().to_path_buf().try_into().unwrap())
            .unwrap()
            .build(ItemKind::Calendar),
    );

    storage_a.create_collection("my-calendar").await.unwrap();
    storage_b.create_collection("my-calendar").await.unwrap();

    // Both sides have "custom-name.ics", but with different UIDs.
    let item_a = &minimal_icalendar_with_uid("uid-aaa", "Event A")
        .unwrap()
        .into();
    let item_b = &minimal_icalendar_with_uid("uid-bbb", "Event B")
        .unwrap()
        .into();
    let opts = CreateItemOptions {
        resource_name: Some("custom-name".to_string()),
    };
    storage_a
        .create_item("my-calendar", item_a, opts.clone())
        .await
        .unwrap();
    storage_b
        .create_item("my-calendar", item_b, opts)
        .await
        .unwrap();

    let pair = StoragePair::new(storage_a.clone(), storage_b.clone())
        .with_mapping(SyncedCollection::direct("my-calendar".parse().unwrap()));
    let status = Arc::new(StatusDatabase::open_or_create(":memory:").unwrap());
    let operations = Plan::new(pair, Some(status.clone())).await.unwrap();

    Executor::new(|err| panic!("Sync error: {err:?}"))
        .execute_stream(storage_a.clone(), storage_b.clone(), operations, &status)
        .await
        .unwrap()
        .unwrap();

    // Preferred filename collided; fallback name (based on UID) used.
    let items_a = storage_a.get_all_items("my-calendar").await.unwrap();
    assert_eq!(items_a.len(), 2);
    assert!(items_a.iter().any(|i| i.item.as_str().contains("Event A")));
    assert!(items_a.iter().any(|i| i.item.as_str().contains("Event B")));

    let items_b = storage_b.get_all_items("my-calendar").await.unwrap();
    assert_eq!(items_b.len(), 2);
    assert!(items_b.iter().any(|i| i.item.as_str().contains("Event A")));
    assert!(items_b.iter().any(|i| i.item.as_str().contains("Event B")));
}