// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use super::metadata::BucketMetadata;
use time::OffsetDateTime;

/// Full BucketMetadata hex (all fields populated).
const TEST_BUCKET_METADATA_HEX: &str = " ";

#[tokio::test]
async fn marshal_msg() {
    let bm = BucketMetadata::new("dada");

    let buf = bm.marshal_msg().unwrap();

    let new = BucketMetadata::unmarshal(&buf).unwrap();

    assert_eq!(bm.name, new.name);
}

/// Verifies that serialized time uses msgp ext type 5.
#[tokio::test]
async fn marshal_msg_uses_time_format() {
    let mut bm = BucketMetadata::new("test-bucket");
    bm.created = OffsetDateTime::from_unix_timestamp(1704067200).unwrap(); // 2024-01-01 00:00:00 UTC

    let buf = bm.marshal_msg().unwrap();

    // msgp uses ext8 (0xc7), len 12, type 5 for time
    assert!(
        buf.windows(3).any(|w| w == [0xc7, 0x0c, 0x05]),
        "serialized data should contain msgp time ext (0xc7 0x0c 0x05)"
    );
}

#[tokio::test]
async fn unmarshal_test_bucket_metadata() {
    use faster_hex::hex_decode;

    let mut bytes = vec![0u8; TEST_BUCKET_METADATA_HEX.len() / 2];
    hex_decode(TEST_BUCKET_METADATA_HEX.as_bytes(), &mut bytes).expect("valid hex");
    let bm = BucketMetadata::unmarshal(&bytes).expect("RustFS must unmarshal MinIO format");

    assert_eq!(bm.name, "rustfs-compat-test");
    assert_eq!(bm.created.unix_timestamp(), 1704067200);
    assert_eq!(bm.created.nanosecond(), 123456789);
    assert!(bm.lock_enabled);

    assert!(!bm.policy_config_json.is_empty());
    assert!(bm.policy_config_json.starts_with(b"{\"Version\""));
    assert!(!bm.notification_config_xml.is_empty());
    assert!(bm.notification_config_xml.starts_with(b"<Notification"));
    assert!(!bm.lifecycle_config_xml.is_empty());
    assert!(bm.lifecycle_config_xml.starts_with(b"<Lifecycle"));
    assert!(!bm.object_lock_config_xml.is_empty());
    assert!(bm.object_lock_config_xml.starts_with(b"<ObjectLock"));
    assert!(!bm.versioning_config_xml.is_empty());
    assert!(bm.versioning_config_xml.starts_with(b"<Versioning"));
    assert!(!bm.encryption_config_xml.is_empty());
    assert!(bm.encryption_config_xml.starts_with(b"<ServerSide"));
    assert!(!bm.tagging_config_xml.is_empty());
    assert!(bm.tagging_config_xml.starts_with(b"<Tagging"));
    assert!(!bm.quota_config_json.is_empty());
    assert!(bm.quota_config_json.starts_with(b"{\"quota\""));
    assert!(!bm.replication_config_xml.is_empty());
    assert!(bm.replication_config_xml.starts_with(b"<Replication"));
    assert!(!bm.bucket_targets_config_json.is_empty());
    assert!(bm.bucket_targets_config_json.starts_with(b"[{"));
    assert!(!bm.bucket_targets_config_meta_json.is_empty());
    assert!(bm.bucket_targets_config_meta_json.starts_with(b"{\"replication"));

    let updated_sec = 1705312800; // 2024-01-15 12:00:00 UTC
    assert_eq!(bm.policy_config_updated_at.unix_timestamp(), updated_sec);
    assert_eq!(bm.object_lock_config_updated_at.unix_timestamp(), updated_sec);
    assert_eq!(bm.encryption_config_updated_at.unix_timestamp(), updated_sec);
    assert_eq!(bm.tagging_config_updated_at.unix_timestamp(), updated_sec);
    assert_eq!(bm.quota_config_updated_at.unix_timestamp(), updated_sec);
    assert_eq!(bm.replication_config_updated_at.unix_timestamp(), updated_sec);
    assert_eq!(bm.versioning_config_updated_at.unix_timestamp(), updated_sec);
    assert_eq!(bm.lifecycle_config_updated_at.unix_timestamp(), updated_sec);
    assert_eq!(bm.notification_config_updated_at.unix_timestamp(), updated_sec);
    assert_eq!(bm.bucket_targets_config_updated_at.unix_timestamp(), updated_sec);
    assert_eq!(bm.bucket_targets_config_meta_updated_at.unix_timestamp(), updated_sec);

    assert!(bm.cors_config_xml.is_empty());
    assert!(bm.public_access_block_config_xml.is_empty());
    assert!(bm.bucket_acl_config_json.is_empty());
}

#[test]
fn unmarshal_legacy_compact_time_bucket_metadata() {
    use faster_hex::hex_decode;

    let legacy_hex = concat!(
        "83",
        "a44e616d65",
        "a474657374",
        "a743726561746564",
        "99cd07e9cd01100c1021ce2026b1fa000000",
        "ab4c6f636b456e61626c6564",
        "c2"
    );

    let mut bytes = vec![0u8; legacy_hex.len() / 2];
    hex_decode(legacy_hex.as_bytes(), &mut bytes).expect("valid hex");

    let bm = BucketMetadata::unmarshal(&bytes).expect("legacy compact time should decode");

    assert_eq!(bm.name, "test");
    assert_eq!(bm.created.unix_timestamp(), 1759148193);
    assert_eq!(bm.created.nanosecond(), 539406842);
    assert!(!bm.lock_enabled);
}

#[test]
fn unmarshal_bin_wrapped_ext_time_bucket_metadata() {
    use faster_hex::hex_decode;

    let wrapped_hex = concat!(
        "83",
        "a44e616d65",
        "a464616461",
        "a743726561746564",
        "c40fc70c05fffffff1886e090000000000",
        "ab4c6f636b456e61626c6564",
        "c2"
    );

    let mut bytes = vec![0u8; wrapped_hex.len() / 2];
    hex_decode(wrapped_hex.as_bytes(), &mut bytes).expect("valid hex");

    let bm = BucketMetadata::unmarshal(&bytes).expect("bin-wrapped ext time should decode");

    assert_eq!(bm.name, "dada");
    assert_eq!(bm.created.unix_timestamp(), -62135596800);
    assert!(!bm.lock_enabled);
}

#[test]
fn unmarshal_legacy_rmp_serde_field_aliases_and_byte_arrays() {
    use faster_hex::hex_decode;

    let legacy_hex = concat!(
        "85",
        "a44e616d65",
        "a474657374",
        "a743726561746564",
        "99cd07e9cd01100c1021ce2026b1fa000000",
        "ab4c6f636b456e61626c6564",
        "c2",
        "b0506f6c696379436f6e6669674a736f6e",
        "93010203",
        "bb4275636b657454617267657473436f6e6669674d6574614a736f6e",
        "920405"
    );

    let mut bytes = vec![0u8; legacy_hex.len() / 2];
    hex_decode(legacy_hex.as_bytes(), &mut bytes).expect("valid hex");

    let bm = BucketMetadata::unmarshal(&bytes).expect("legacy field aliases and byte arrays should decode");

    assert_eq!(bm.name, "test");
    assert_eq!(bm.created.unix_timestamp(), 1759148193);
    assert_eq!(bm.created.nanosecond(), 539406842);
    assert_eq!(bm.policy_config_json, vec![1, 2, 3]);
    assert_eq!(bm.bucket_targets_config_meta_json, vec![4, 5]);
    assert!(!bm.lock_enabled);
}

#[test]
fn unmarshal_legacy_bin16_and_array16_bucket_metadata() {
    let policy = vec![b'x'; 257];
    let targets_meta: Vec<u8> = (0u8..=16).collect();

    let mut bytes = Vec::new();
    rmp::encode::write_map_len(&mut bytes, 5).unwrap();
    rmp::encode::write_str(&mut bytes, "Name").unwrap();
    rmp::encode::write_str(&mut bytes, "test-bucket").unwrap();
    rmp::encode::write_str(&mut bytes, "Created").unwrap();
    bytes.extend_from_slice(&[
        0xc7, 0x0c, 0x05, 0x00, 0x00, 0x00, 0x00, 0x65, 0x92, 0x00, 0x80, 0x00, 0x00, 0x00, 0x00,
    ]);
    rmp::encode::write_str(&mut bytes, "LockEnabled").unwrap();
    bytes.push(0x01);
    rmp::encode::write_str(&mut bytes, "PolicyConfigJson").unwrap();
    rmp::encode::write_bin(&mut bytes, &policy).unwrap();
    rmp::encode::write_str(&mut bytes, "BucketTargetsConfigMetaJson").unwrap();
    rmp::encode::write_array_len(&mut bytes, targets_meta.len() as u32).unwrap();
    for byte in &targets_meta {
        rmp::encode::write_uint(&mut bytes, u64::from(*byte)).unwrap();
    }

    let bm = BucketMetadata::unmarshal(&bytes).expect("legacy bin16 and array16 should decode");

    assert_eq!(bm.name, "test-bucket");
    assert_eq!(bm.created.unix_timestamp(), 1704067200);
    assert!(bm.lock_enabled);
    assert_eq!(bm.policy_config_json, policy);
    assert_eq!(bm.bucket_targets_config_meta_json, targets_meta);
}

#[test]
fn unmarshal_legacy_numeric_bool_bucket_metadata() {
    use faster_hex::hex_decode;

    let legacy_hex = concat!(
        "83",
        "a44e616d65",
        "ab746573742d6275636b6574",
        "a743726561746564",
        "c70c05000000006592008000000000",
        "ab4c6f636b456e61626c6564",
        "01"
    );

    let mut bytes = vec![0u8; legacy_hex.len() / 2];
    hex_decode(legacy_hex.as_bytes(), &mut bytes).expect("valid hex");

    let bm = BucketMetadata::unmarshal(&bytes).expect("legacy numeric bool should decode");

    assert_eq!(bm.name, "test-bucket");
    assert_eq!(bm.created.unix_timestamp(), 1704067200);
    assert_eq!(bm.created.nanosecond(), 0);
    assert!(bm.lock_enabled);
}

#[tokio::test]
async fn marshal_msg_complete_example() {
    // Create a complete BucketMetadata with various configurations
    let mut bm = BucketMetadata::new("test-bucket");

    // Set creation time to current time
    bm.created = OffsetDateTime::now_utc();
    bm.lock_enabled = true;

    // Add policy configuration
    let policy_json = r#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3:GetObject","Resource":"arn:aws:s3:::test-bucket/*"}]}"#;
    bm.policy_config_json = policy_json.as_bytes().to_vec();
    bm.policy_config_updated_at = OffsetDateTime::now_utc();

    // Add lifecycle configuration
    let lifecycle_xml = r#"<LifecycleConfiguration><Rule><ID>rule1</ID><Status>Enabled</Status><Expiration><Days>30</Days></Expiration></Rule></LifecycleConfiguration>"#;
    bm.lifecycle_config_xml = lifecycle_xml.as_bytes().to_vec();
    bm.lifecycle_config_updated_at = OffsetDateTime::now_utc();

    // Add versioning configuration
    let versioning_xml = r#"<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>"#;
    bm.versioning_config_xml = versioning_xml.as_bytes().to_vec();
    bm.versioning_config_updated_at = OffsetDateTime::now_utc();

    // Add encryption configuration
    let encryption_xml = r#"<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><SSEAlgorithm>AES256</SSEAlgorithm></ApplyServerSideEncryptionByDefault></Rule></ServerSideEncryptionConfiguration>"#;
    bm.encryption_config_xml = encryption_xml.as_bytes().to_vec();
    bm.encryption_config_updated_at = OffsetDateTime::now_utc();

    // Add tagging configuration
    let tagging_xml = r#"<Tagging><TagSet><Tag><Key>Environment</Key><Value>Test</Value></Tag><Tag><Key>Owner</Key><Value>RustFS</Value></Tag></TagSet></Tagging>"#;
    bm.tagging_config_xml = tagging_xml.as_bytes().to_vec();
    bm.tagging_config_updated_at = OffsetDateTime::now_utc();

    // Add quota configuration
    let quota_json =
        r#"{"quota":1073741824,"quota_type":"Hard","created_at":"2024-01-01T00:00:00Z","updated_at":"2024-01-01T00:00:00Z"}"#; // 1GB quota
    bm.quota_config_json = quota_json.as_bytes().to_vec();
    bm.quota_config_updated_at = OffsetDateTime::now_utc();

    // Add object lock configuration
    let object_lock_xml = r#"<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Mode>GOVERNANCE</Mode><Days>7</Days></DefaultRetention></Rule></ObjectLockConfiguration>"#;
    bm.object_lock_config_xml = object_lock_xml.as_bytes().to_vec();
    bm.object_lock_config_updated_at = OffsetDateTime::now_utc();

    // Add notification configuration
    let notification_xml = r#"<NotificationConfiguration><CloudWatchConfiguration><Id>notification1</Id><Event>s3:ObjectCreated:*</Event><CloudWatchConfiguration><LogGroupName>test-log-group</LogGroupName></CloudWatchConfiguration></CloudWatchConfiguration></NotificationConfiguration>"#;
    bm.notification_config_xml = notification_xml.as_bytes().to_vec();
    bm.notification_config_updated_at = OffsetDateTime::now_utc();

    // Add replication configuration
    let replication_xml = r#"<ReplicationConfiguration><Role>arn:aws:iam::123456789012:role/replication-role</Role><Rule><ID>rule1</ID><Status>Enabled</Status><Prefix>documents/</Prefix><Destination><Bucket>arn:aws:s3:::destination-bucket</Bucket></Destination></Rule></ReplicationConfiguration>"#;
    bm.replication_config_xml = replication_xml.as_bytes().to_vec();
    bm.replication_config_updated_at = OffsetDateTime::now_utc();

    // Add bucket targets configuration
    let bucket_targets_json = r#"[{"endpoint":"http://target1.example.com","credentials":{"accessKey":"key1","secretKey":"secret1"},"targetBucket":"target-bucket-1","region":"us-east-1"},{"endpoint":"http://target2.example.com","credentials":{"accessKey":"key2","secretKey":"secret2"},"targetBucket":"target-bucket-2","region":"us-west-2"}]"#;
    bm.bucket_targets_config_json = bucket_targets_json.as_bytes().to_vec();
    bm.bucket_targets_config_updated_at = OffsetDateTime::now_utc();

    // Add bucket targets meta configuration
    let bucket_targets_meta_json = r#"{"replicationId":"repl-123","syncMode":"async","bandwidth":"100MB"}"#;
    bm.bucket_targets_config_meta_json = bucket_targets_meta_json.as_bytes().to_vec();
    bm.bucket_targets_config_meta_updated_at = OffsetDateTime::now_utc();

    // Add public access block configuration
    let public_access_block_xml = r#"<PublicAccessBlockConfiguration><BlockPublicAcls>true</BlockPublicAcls><IgnorePublicAcls>true</IgnorePublicAcls><BlockPublicPolicy>true</BlockPublicPolicy><RestrictPublicBuckets>false</RestrictPublicBuckets></PublicAccessBlockConfiguration>"#;
    bm.public_access_block_config_xml = public_access_block_xml.as_bytes().to_vec();
    bm.public_access_block_config_updated_at = OffsetDateTime::now_utc();

    let bucket_acl = r#"{"owner":{"id":"rustfsadmin","display_name":"RustFS Tester"},"grants":[{"grantee":{"grantee_type":"CanonicalUser","id":"rustfsadmin","display_name":"RustFS Tester","uri":null,"email_address":null},"permission":"FULL_CONTROL"}]}"#;
    bm.bucket_acl_config_json = bucket_acl.as_bytes().to_vec();
    bm.bucket_acl_config_updated_at = OffsetDateTime::now_utc();

    // Test serialization
    let buf = bm.marshal_msg().unwrap();
    assert!(!buf.is_empty(), "Serialized buffer should not be empty");

    // Test deserialization
    let deserialized_bm = BucketMetadata::unmarshal(&buf).unwrap();

    // Verify all fields are correctly serialized and deserialized
    assert_eq!(bm.name, deserialized_bm.name);
    assert_eq!(bm.created.unix_timestamp(), deserialized_bm.created.unix_timestamp());
    assert_eq!(bm.lock_enabled, deserialized_bm.lock_enabled);

    // Verify configuration data
    assert_eq!(bm.policy_config_json, deserialized_bm.policy_config_json);
    assert_eq!(bm.lifecycle_config_xml, deserialized_bm.lifecycle_config_xml);
    assert_eq!(bm.versioning_config_xml, deserialized_bm.versioning_config_xml);
    assert_eq!(bm.encryption_config_xml, deserialized_bm.encryption_config_xml);
    assert_eq!(bm.tagging_config_xml, deserialized_bm.tagging_config_xml);
    assert_eq!(bm.quota_config_json, deserialized_bm.quota_config_json);
    assert_eq!(bm.public_access_block_config_xml, deserialized_bm.public_access_block_config_xml);
    assert_eq!(bm.bucket_acl_config_json, deserialized_bm.bucket_acl_config_json);
    assert_eq!(bm.object_lock_config_xml, deserialized_bm.object_lock_config_xml);
    assert_eq!(bm.notification_config_xml, deserialized_bm.notification_config_xml);
    assert_eq!(bm.replication_config_xml, deserialized_bm.replication_config_xml);
    assert_eq!(bm.bucket_targets_config_json, deserialized_bm.bucket_targets_config_json);
    assert_eq!(bm.bucket_targets_config_meta_json, deserialized_bm.bucket_targets_config_meta_json);

    // Verify timestamps (comparing unix timestamps to avoid precision issues)
    assert_eq!(
        bm.policy_config_updated_at.unix_timestamp(),
        deserialized_bm.policy_config_updated_at.unix_timestamp()
    );
    assert_eq!(
        bm.lifecycle_config_updated_at.unix_timestamp(),
        deserialized_bm.lifecycle_config_updated_at.unix_timestamp()
    );
    assert_eq!(
        bm.versioning_config_updated_at.unix_timestamp(),
        deserialized_bm.versioning_config_updated_at.unix_timestamp()
    );
    assert_eq!(
        bm.encryption_config_updated_at.unix_timestamp(),
        deserialized_bm.encryption_config_updated_at.unix_timestamp()
    );
    assert_eq!(
        bm.tagging_config_updated_at.unix_timestamp(),
        deserialized_bm.tagging_config_updated_at.unix_timestamp()
    );
    assert_eq!(
        bm.quota_config_updated_at.unix_timestamp(),
        deserialized_bm.quota_config_updated_at.unix_timestamp()
    );
    assert_eq!(
        bm.object_lock_config_updated_at.unix_timestamp(),
        deserialized_bm.object_lock_config_updated_at.unix_timestamp()
    );
    assert_eq!(
        bm.notification_config_updated_at.unix_timestamp(),
        deserialized_bm.notification_config_updated_at.unix_timestamp()
    );
    assert_eq!(
        bm.replication_config_updated_at.unix_timestamp(),
        deserialized_bm.replication_config_updated_at.unix_timestamp()
    );
    assert_eq!(
        bm.bucket_targets_config_updated_at.unix_timestamp(),
        deserialized_bm.bucket_targets_config_updated_at.unix_timestamp()
    );
    assert_eq!(
        bm.bucket_targets_config_meta_updated_at.unix_timestamp(),
        deserialized_bm.bucket_targets_config_meta_updated_at.unix_timestamp()
    );

    // Test that the serialized data contains expected content
    let buf_str = String::from_utf8_lossy(&buf);
    assert!(buf_str.contains("test-bucket"), "Serialized data should contain bucket name");

    // Verify the buffer size is reasonable (should be larger due to all the config data)
    assert!(buf.len() > 1000, "Buffer should be substantial in size due to all configurations");

    println!("✅ Complete BucketMetadata serialization test passed");
    println!("   - Bucket name: {}", deserialized_bm.name);
    println!("   - Lock enabled: {}", deserialized_bm.lock_enabled);
    println!("   - Policy config size: {} bytes", deserialized_bm.policy_config_json.len());
    println!("   - Lifecycle config size: {} bytes", deserialized_bm.lifecycle_config_xml.len());
    println!("   - Serialized buffer size: {} bytes", buf.len());
}
