use std::time::Duration;

use s2_common::{
    maybe::Maybe,
    types::{
        basin::{BasinNamePrefix, BasinNameStartAfter, ListBasinsRequest},
        config::{
            BasinConfig, BasinReconfiguration, OptionalDeleteOnEmptyConfig, OptionalStreamConfig,
            RetentionPolicy, StorageClass, StreamReconfiguration, TimestampingMode,
            TimestampingReconfiguration,
        },
        resources::{ListItemsRequestParts, ProvisionMode, ProvisionResult, RequestToken},
    },
};
use s2_lite::backend::error::{
    DeleteBasinError, GetBasinConfigError, ProvisionBasinError, ReconfigureBasinError,
};

use super::common::*;

#[tokio::test]
async fn test_create_basin_idempotency_respects_request_token() {
    let backend = create_backend().await;
    let basin_name = test_basin_name("basin-idempotency");
    let config = BasinConfig {
        create_stream_on_append: true,
        ..Default::default()
    };

    let token1: RequestToken = "token-1".parse().unwrap();

    let created = backend
        .provision_basin(
            basin_name.clone(),
            config.clone(),
            ProvisionMode::CreateOnly {
                request_token: Some(token1.clone()),
            },
        )
        .await
        .expect("Failed to create basin");
    assert!(matches!(
        created,
        ProvisionResult::Created(ref info) if info.deleted_at.is_none()
            && info.created_at <= time::OffsetDateTime::now_utc()
    ));

    let idempotent = backend
        .provision_basin(
            basin_name.clone(),
            config.clone(),
            ProvisionMode::CreateOnly {
                request_token: Some(token1.clone()),
            },
        )
        .await
        .expect("Idempotent create should succeed with same request token");
    assert!(matches!(
        idempotent,
        ProvisionResult::Noop(ref info) if info.deleted_at.is_none()
            && info.created_at <= time::OffsetDateTime::now_utc()
    ));

    let different_token: RequestToken = "token-2".parse().unwrap();
    let different_token_result = backend
        .provision_basin(
            basin_name.clone(),
            config.clone(),
            ProvisionMode::CreateOnly {
                request_token: Some(different_token),
            },
        )
        .await;
    assert!(matches!(
        different_token_result,
        Err(ProvisionBasinError::BasinAlreadyExists(_))
    ));

    let mut different_config = config.clone();
    different_config.create_stream_on_append = false;
    let different_config_result = backend
        .provision_basin(
            basin_name,
            different_config,
            ProvisionMode::CreateOnly {
                request_token: Some(token1),
            },
        )
        .await;
    assert!(matches!(
        different_config_result,
        Err(ProvisionBasinError::BasinAlreadyExists(_))
    ));
}

#[tokio::test]
async fn test_ensure_preserves_idempotency_key() {
    let backend = create_backend().await;
    let basin_name = test_basin_name("idempotency-key-preserve");
    let config = BasinConfig {
        create_stream_on_append: true,
        ..Default::default()
    };

    let token: RequestToken = "my-request-token".parse().unwrap();

    backend
        .provision_basin(
            basin_name.clone(),
            config.clone(),
            ProvisionMode::CreateOnly {
                request_token: Some(token.clone()),
            },
        )
        .await
        .expect("Failed to create basin");

    backend
        .provision_basin(
            basin_name.clone(),
            config.clone(),
            ProvisionMode::CreateOnly {
                request_token: Some(token.clone()),
            },
        )
        .await
        .expect("Idempotency should work before Ensure");

    let mut updated_config = config.clone();
    updated_config.create_stream_on_read = true;
    backend
        .provision_basin(basin_name.clone(), updated_config, ProvisionMode::Ensure)
        .await
        .expect("Ensure should succeed");

    backend
        .provision_basin(
            basin_name.clone(),
            config.clone(),
            ProvisionMode::CreateOnly {
                request_token: Some(token.clone()),
            },
        )
        .await
        .expect("Idempotency should still work after Ensure");
}

#[tokio::test]
async fn test_provision_basin_ensure_updates_config() {
    let backend = create_backend().await;
    let basin_name = test_basin_name("basin-recreate");
    let initial_config = BasinConfig {
        create_stream_on_append: false,
        create_stream_on_read: false,
        ..Default::default()
    };

    backend
        .provision_basin(
            basin_name.clone(),
            initial_config.clone(),
            ProvisionMode::CreateOnly {
                request_token: None,
            },
        )
        .await
        .expect("Failed to create basin");

    let mut updated_config = initial_config.clone();
    updated_config.create_stream_on_append = true;
    updated_config.create_stream_on_read = true;
    updated_config.default_stream_config.storage_class = Some(StorageClass::Standard);

    backend
        .provision_basin(
            basin_name.clone(),
            updated_config.clone(),
            ProvisionMode::Ensure,
        )
        .await
        .expect("Ensure should update basin config");

    let stored_config = backend
        .get_basin_config(basin_name.clone())
        .await
        .expect("Failed to fetch basin config");
    assert!(stored_config.create_stream_on_append);
    assert!(stored_config.create_stream_on_read);
    assert_eq!(
        stored_config.default_stream_config.storage_class,
        Some(StorageClass::Standard)
    );

    backend
        .provision_basin(
            basin_name.clone(),
            updated_config,
            ProvisionMode::CreateOnly {
                request_token: None,
            },
        )
        .await
        .expect_err("CreateOnly without request token should not be idempotent");
}

#[tokio::test]
async fn test_provision_basin_ensure_resets_unspecified_config() {
    let backend = create_backend().await;
    let basin_name = test_basin_name("basin-ensure-reset");
    let initial_config = BasinConfig {
        create_stream_on_append: true,
        create_stream_on_read: false,
        default_stream_config: OptionalStreamConfig {
            storage_class: Some(StorageClass::Standard),
            retention_policy: Some(RetentionPolicy::Infinite()),
            ..Default::default()
        },
        ..Default::default()
    };

    backend
        .provision_basin(
            basin_name.clone(),
            initial_config,
            ProvisionMode::CreateOnly {
                request_token: None,
            },
        )
        .await
        .expect("Failed to create basin");

    backend
        .provision_basin(
            basin_name.clone(),
            BasinConfig {
                create_stream_on_read: true,
                ..Default::default()
            },
            ProvisionMode::Ensure,
        )
        .await
        .expect("Ensure should reset unspecified fields to defaults");

    let stored_config = backend
        .get_basin_config(basin_name)
        .await
        .expect("Failed to fetch basin config");
    assert!(!stored_config.create_stream_on_append);
    assert!(stored_config.create_stream_on_read);
    assert_eq!(stored_config.default_stream_config.storage_class, None);
    assert_eq!(stored_config.default_stream_config.retention_policy, None);
}

#[tokio::test]
async fn test_provision_basin_ensure_noops_with_explicit_zero_delete_on_empty() {
    let backend = create_backend().await;
    let basin_name = test_basin_name("basin-zero-doe-noop");
    let config = BasinConfig {
        default_stream_config: OptionalStreamConfig {
            delete_on_empty: OptionalDeleteOnEmptyConfig {
                min_age: Some(Duration::ZERO),
            },
            ..Default::default()
        },
        ..Default::default()
    };

    backend
        .provision_basin(
            basin_name.clone(),
            config.clone(),
            ProvisionMode::CreateOnly {
                request_token: None,
            },
        )
        .await
        .expect("Failed to create basin");

    let ensured = backend
        .provision_basin(basin_name.clone(), config, ProvisionMode::Ensure)
        .await
        .expect("Ensure should succeed");

    assert!(matches!(ensured, ProvisionResult::Noop(_)));

    let stored_config = backend
        .get_basin_config(basin_name)
        .await
        .expect("Failed to fetch basin config");
    assert_eq!(
        stored_config.default_stream_config.delete_on_empty.min_age,
        Some(Duration::ZERO)
    );
}

#[tokio::test]
async fn test_reconfigure_basin_updates_nested_defaults() {
    let backend = create_backend().await;
    let basin_name = test_basin_name("basin-reconfigure");
    let mut initial_config = BasinConfig::default();
    initial_config.default_stream_config.storage_class = Some(StorageClass::Standard);

    backend
        .provision_basin(
            basin_name.clone(),
            initial_config.clone(),
            ProvisionMode::CreateOnly {
                request_token: None,
            },
        )
        .await
        .expect("Failed to create basin");

    let timestamping_reconfig = TimestampingReconfiguration {
        mode: Maybe::from(Some(TimestampingMode::Arrival)),
        ..Default::default()
    };
    let mut stream_reconfig = StreamReconfiguration {
        storage_class: Maybe::from(Some(StorageClass::Express)),
        retention_policy: Maybe::from(Some(RetentionPolicy::Infinite())),
        ..Default::default()
    };
    stream_reconfig.timestamping = Maybe::from(Some(timestamping_reconfig));

    let reconfig = BasinReconfiguration {
        default_stream_config: Maybe::from(Some(stream_reconfig)),
        stream_cipher: Maybe::default(),
        create_stream_on_append: Maybe::from(true),
        create_stream_on_read: Maybe::from(true),
    };

    let updated = backend
        .reconfigure_basin(basin_name.clone(), reconfig)
        .await
        .expect("Failed to reconfigure basin");

    assert!(updated.create_stream_on_append);
    assert!(updated.create_stream_on_read);
    assert_eq!(
        updated.default_stream_config.storage_class,
        Some(StorageClass::Express)
    );
    assert_eq!(
        updated.default_stream_config.retention_policy,
        Some(RetentionPolicy::Infinite())
    );
    assert_eq!(
        updated.default_stream_config.timestamping.mode,
        Some(TimestampingMode::Arrival)
    );

    let fetched = backend
        .get_basin_config(basin_name)
        .await
        .expect("Failed to fetch basin config after reconfigure");
    assert_eq!(
        fetched.default_stream_config.storage_class,
        Some(StorageClass::Express)
    );
    assert_eq!(
        fetched.default_stream_config.retention_policy,
        Some(RetentionPolicy::Infinite())
    );
    assert_eq!(
        fetched.default_stream_config.timestamping.mode,
        Some(TimestampingMode::Arrival)
    );
    assert!(fetched.create_stream_on_append);
    assert!(fetched.create_stream_on_read);
}

#[tokio::test]
async fn test_delete_basin_marks_deleting_and_blocks_create() {
    let backend = create_backend().await;
    let basin_name = test_basin_name("basin-delete");

    backend
        .provision_basin(
            basin_name.clone(),
            BasinConfig::default(),
            ProvisionMode::CreateOnly {
                request_token: None,
            },
        )
        .await
        .expect("Failed to create basin");

    backend
        .delete_basin(basin_name.clone())
        .await
        .expect("Failed to delete basin");

    let page = backend
        .list_basins(ListBasinsRequest::default())
        .await
        .expect("Failed to list basins");
    let info = page
        .values
        .iter()
        .find(|info| info.name == basin_name)
        .expect("Deleted basin should appear in listing");
    assert!(info.deleted_at.is_some());

    let reconfigure_result = backend
        .reconfigure_basin(basin_name.clone(), BasinReconfiguration::default())
        .await;
    assert!(matches!(
        reconfigure_result,
        Err(ReconfigureBasinError::BasinDeletionPending(_))
    ));

    let recreate_result = backend
        .provision_basin(
            basin_name.clone(),
            BasinConfig::default(),
            ProvisionMode::CreateOnly {
                request_token: None,
            },
        )
        .await;
    assert!(matches!(
        recreate_result,
        Err(ProvisionBasinError::BasinDeletionPending(_))
    ));

    backend
        .delete_basin(basin_name)
        .await
        .expect("Second delete should be idempotent");
}

#[tokio::test]
async fn test_get_nonexistent_basin_config() {
    let backend = create_backend().await;
    let basin_name = test_basin_name("nonexistent-basin");

    let result = backend.get_basin_config(basin_name).await;

    assert!(matches!(result, Err(GetBasinConfigError::BasinNotFound(_))));
}

#[tokio::test]
async fn test_delete_nonexistent_basin_returns_not_found() {
    let backend = create_backend().await;
    let basin_name = test_basin_name("delete-missing-basin");

    let result = backend.delete_basin(basin_name).await;

    assert!(matches!(result, Err(DeleteBasinError::BasinNotFound(_))));
}

#[tokio::test]
async fn test_list_basins_empty() {
    let backend = create_backend().await;

    let page = backend
        .list_basins(ListBasinsRequest::default())
        .await
        .expect("Failed to list basins");

    assert!(page.values.is_empty());
    assert!(!page.has_more);
}

#[tokio::test]
async fn test_list_basins_multiple() {
    let backend = create_backend().await;

    for i in 0..5 {
        create_test_basin(&backend, &format!("list-{}", i), BasinConfig::default()).await;
    }

    let page = backend
        .list_basins(ListBasinsRequest::default())
        .await
        .expect("Failed to list basins");

    let names: Vec<_> = page.values.iter().map(|info| info.name.as_ref()).collect();
    assert_eq!(
        names,
        vec![
            "test-basin-list-0",
            "test-basin-list-1",
            "test-basin-list-2",
            "test-basin-list-3",
            "test-basin-list-4",
        ]
    );
    assert!(!page.has_more);
}

#[tokio::test]
async fn test_list_basins_pagination() {
    let backend = create_backend().await;

    for i in 0..15 {
        create_test_basin(
            &backend,
            &format!("paginated-{:02}", i),
            BasinConfig::default(),
        )
        .await;
    }

    let page1 = backend
        .list_basins(
            ListItemsRequestParts {
                prefix: BasinNamePrefix::default(),
                start_after: BasinNameStartAfter::default(),
                limit: 5.into(),
            }
            .try_into()
            .unwrap(),
        )
        .await
        .expect("Failed to list basins page 1");

    assert!(page1.has_more);
    let page1_names: Vec<_> = page1.values.iter().map(|info| info.name.as_ref()).collect();
    assert_eq!(
        page1_names,
        vec![
            "test-basin-paginated-00",
            "test-basin-paginated-01",
            "test-basin-paginated-02",
            "test-basin-paginated-03",
            "test-basin-paginated-04",
        ]
    );

    let page2 = backend
        .list_basins(
            ListItemsRequestParts {
                prefix: BasinNamePrefix::default(),
                start_after: page1.values.last().unwrap().name.clone().into(),
                limit: 5.into(),
            }
            .try_into()
            .unwrap(),
        )
        .await
        .expect("Failed to list basins page 2");

    assert!(page2.has_more);
    let page2_names: Vec<_> = page2.values.iter().map(|info| info.name.as_ref()).collect();
    assert_eq!(
        page2_names,
        vec![
            "test-basin-paginated-05",
            "test-basin-paginated-06",
            "test-basin-paginated-07",
            "test-basin-paginated-08",
            "test-basin-paginated-09",
        ]
    );

    let page3 = backend
        .list_basins(
            ListItemsRequestParts {
                prefix: BasinNamePrefix::default(),
                start_after: page2.values.last().unwrap().name.clone().into(),
                limit: 5.into(),
            }
            .try_into()
            .unwrap(),
        )
        .await
        .expect("Failed to list basins page 3");

    assert!(!page3.has_more);
    let page3_names: Vec<_> = page3.values.iter().map(|info| info.name.as_ref()).collect();
    assert_eq!(
        page3_names,
        vec![
            "test-basin-paginated-10",
            "test-basin-paginated-11",
            "test-basin-paginated-12",
            "test-basin-paginated-13",
            "test-basin-paginated-14",
        ]
    );
}

#[tokio::test]
async fn test_list_basins_prefix_filter() {
    let backend = create_backend().await;

    create_test_basin(&backend, "prod-app-1", BasinConfig::default()).await;
    create_test_basin(&backend, "prod-app-2", BasinConfig::default()).await;
    create_test_basin(&backend, "dev-app-1", BasinConfig::default()).await;
    create_test_basin(&backend, "staging-app-1", BasinConfig::default()).await;

    let prod_basins = backend
        .list_basins(
            ListItemsRequestParts {
                prefix: "test-basin-prod-".parse().unwrap(),
                start_after: BasinNameStartAfter::default(),
                limit: Default::default(),
            }
            .try_into()
            .unwrap(),
        )
        .await
        .expect("Failed to list basins with prefix");

    let prod_names: Vec<_> = prod_basins
        .values
        .iter()
        .map(|info| info.name.as_ref())
        .collect();
    assert_eq!(
        prod_names,
        vec!["test-basin-prod-app-1", "test-basin-prod-app-2"]
    );
}

#[tokio::test]
async fn test_list_basins_prefix_with_pagination() {
    let backend = create_backend().await;

    for i in 0..10 {
        create_test_basin(
            &backend,
            &format!("prefixed-{:02}", i),
            BasinConfig::default(),
        )
        .await;
    }
    create_test_basin(&backend, "other-basin", BasinConfig::default()).await;

    let page1 = backend
        .list_basins(
            ListItemsRequestParts {
                prefix: "test-basin-prefixed-".parse().unwrap(),
                start_after: BasinNameStartAfter::default(),
                limit: 4.into(),
            }
            .try_into()
            .unwrap(),
        )
        .await
        .expect("Failed to list basins");

    assert!(page1.has_more);
    let page1_names: Vec<_> = page1.values.iter().map(|info| info.name.as_ref()).collect();
    assert_eq!(
        page1_names,
        vec![
            "test-basin-prefixed-00",
            "test-basin-prefixed-01",
            "test-basin-prefixed-02",
            "test-basin-prefixed-03",
        ]
    );

    let page2 = backend
        .list_basins(
            ListItemsRequestParts {
                prefix: "test-basin-prefixed-".parse().unwrap(),
                start_after: page1.values.last().unwrap().name.clone().into(),
                limit: 4.into(),
            }
            .try_into()
            .unwrap(),
        )
        .await
        .expect("Failed to list basins");

    assert!(page2.has_more);
    let page2_names: Vec<_> = page2.values.iter().map(|info| info.name.as_ref()).collect();
    assert_eq!(
        page2_names,
        vec![
            "test-basin-prefixed-04",
            "test-basin-prefixed-05",
            "test-basin-prefixed-06",
            "test-basin-prefixed-07",
        ]
    );

    let page3 = backend
        .list_basins(
            ListItemsRequestParts {
                prefix: "test-basin-prefixed-".parse().unwrap(),
                start_after: page2.values.last().unwrap().name.clone().into(),
                limit: 4.into(),
            }
            .try_into()
            .unwrap(),
        )
        .await
        .expect("Failed to list basins");

    assert!(!page3.has_more);
    let page3_names: Vec<_> = page3.values.iter().map(|info| info.name.as_ref()).collect();
    assert_eq!(
        page3_names,
        vec!["test-basin-prefixed-08", "test-basin-prefixed-09"]
    );
}
