//! The cloud-target blob store against a REAL S3-compatible server
//! (MinIO in a container) — same contract the local store satisfies.

use testcontainers_modules::testcontainers::core::{ExecCommand, IntoContainerPort, WaitFor};
use testcontainers_modules::testcontainers::runners::AsyncRunner;
use testcontainers_modules::testcontainers::{GenericImage, ImageExt};

use cm_files::{BlobError, BlobStore, S3BlobStore};

async fn minio_store() -> (
    S3BlobStore,
    testcontainers_modules::testcontainers::ContainerAsync<GenericImage>,
) {
    let container = GenericImage::new("minio/minio", "latest")
        .with_exposed_port(9000.tcp())
        .with_wait_for(WaitFor::message_on_either_std("API:"))
        .with_env_var("MINIO_ROOT_USER", "tc-access")
        .with_env_var("MINIO_ROOT_PASSWORD", "tc-secret-key")
        .with_cmd(["server", "/data"])
        .start()
        .await
        .expect("minio starts");
    // Create the bucket with the bundled mc client.
    container
        .exec(ExecCommand::new([
            "sh",
            "-c",
            "mc alias set local http://127.0.0.1:9000 tc-access tc-secret-key              && mc mb local/clawmates",
        ]))
        .await
        .expect("bucket created");
    let port = container.get_host_port_ipv4(9000).await.unwrap();
    // The bucket is created asynchronously after boot; retry connect+probe.
    let store = S3BlobStore::connect(
        &format!("http://127.0.0.1:{port}"),
        "clawmates",
        "tc-access",
        "tc-secret-key",
    )
    .expect("client builds");
    for _ in 0..50 {
        if store.put("probe", b"x").await.is_ok() {
            store.delete("probe").await.ok();
            return (store, container);
        }
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    }
    panic!("minio bucket never became writable");
}

#[tokio::test]
async fn s3_round_trip_overwrite_and_missing_keys() {
    let (store, _container) = minio_store().await;

    store
        .put("ws1/documents/agent1/report.md", b"# Q2 Report")
        .await
        .unwrap();
    assert_eq!(
        store.get("ws1/documents/agent1/report.md").await.unwrap(),
        b"# Q2 Report"
    );

    // Overwrite replaces.
    store
        .put("ws1/documents/agent1/report.md", b"# Q3 Report")
        .await
        .unwrap();
    assert_eq!(
        store.get("ws1/documents/agent1/report.md").await.unwrap(),
        b"# Q3 Report"
    );

    // Delete then NotFound on both get and delete.
    store
        .delete("ws1/documents/agent1/report.md")
        .await
        .unwrap();
    assert!(matches!(
        store.get("ws1/documents/agent1/report.md").await,
        Err(BlobError::NotFound)
    ));
    assert!(matches!(
        store.delete("ws1/documents/agent1/report.md").await,
        Err(BlobError::NotFound)
    ));

    // Nested keys work without directory semantics.
    store.put("a/b/c/deep.txt", b"deep").await.unwrap();
    assert_eq!(store.get("a/b/c/deep.txt").await.unwrap(), b"deep");
}