use std::{
    io,
    sync::{Arc, Mutex},
};

use admin_api::service::{OpenApiImportCreatePayload, OpenApiUpload, OpenApiUploadLocale};
use crank_artifacts::MAX_ARTIFACT_BYTES;
use crank_core::{Workspace, WorkspaceId, WorkspaceStatus};
use crank_registry::{ArtifactSourceId, ArtifactSourceLifecycle, CreateWorkspaceRequest};
use metrics_util::debugging::DebuggingRecorder;
use opentelemetry::trace::TracerProvider as _;
use opentelemetry_sdk::{
    error::OTelSdkResult,
    trace::{SdkTracerProvider, SpanData, SpanExporter},
};
use reqwest::multipart::{Form, Part};
use serde_json::Value;
use serial_test::serial;
use time::{Duration, OffsetDateTime};
use tracing_subscriber::{fmt::MakeWriter, layer::SubscriberExt};

use super::common::{
    authorized_client, build_test_app, spawn_admin_api, test_auth_settings, test_registry,
    test_secret_crypto, test_service, test_storage_root,
};

mod apply_failures;

const OPENAPI: &str = r#"
openapi: 3.0.3
info: { title: Source authority }
servers:
  - url: https://example.test
paths:
  /health:
    get:
      operationId: sourceHealth
      responses:
        '200': { description: OK }
"#;

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn multipart_accepts_valid_yaml_and_json_through_the_outer_router() {
    let registry = test_registry().await;
    let app = build_test_app(registry, test_storage_root("openapi_valid_multipart"));
    let server = spawn_admin_api(app).await;
    let client = authorized_client(&server).await;

    for (document, filename, mime_type) in [
        (OPENAPI.as_bytes(), "openapi.yaml", "application/yaml"),
        (
            br#"{"openapi":"3.0.3","info":{"title":"JSON"},"paths":{"/health":{"get":{"responses":{"200":{"description":"OK"}}}}}}"#
                .as_slice(),
            "openapi.json",
            "application/json",
        ),
    ] {
        let response = client
            .post(format!("{server}/imports/openapi/preview"))
            .multipart(Form::new().part(
                "file",
                file_part(document, filename, mime_type),
            ))
            .send()
            .await
            .unwrap();
        let status = response.status();
        let body = response.json::<Value>().await.unwrap();
        assert_eq!(status, reqwest::StatusCode::OK, "{body}");
        assert!(body["job_id"].is_string());
        assert!(!body.to_string().contains("source_id"));
        assert!(!body.to_string().contains("digest"));
    }
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn multipart_requires_an_owner_membership_before_reading_the_file() {
    let registry = test_registry().await;
    let now = OffsetDateTime::now_utc();
    let foreign_workspace = WorkspaceId::new("ws_openapi_unauthorized");
    registry
        .create_workspace(CreateWorkspaceRequest {
            workspace: &Workspace {
                id: foreign_workspace.clone(),
                slug: "openapi-unauthorized".to_owned(),
                display_name: "OpenAPI Unauthorized".to_owned(),
                status: WorkspaceStatus::Active,
                settings: serde_json::json!({}),
                created_at: now,
                updated_at: now,
            },
        })
        .await
        .unwrap();
    let app = build_test_app(
        registry.clone(),
        test_storage_root("openapi_upload_authorization"),
    );
    let server = spawn_admin_api(app).await;
    let endpoint = format!("{server}/imports/openapi/preview");

    let anonymous = reqwest::Client::new()
        .post(&endpoint)
        .header("content-type", "multipart/form-data; boundary=broken")
        .body("this body must not be parsed before authentication")
        .send()
        .await
        .unwrap();
    assert_eq!(anonymous.status(), reqwest::StatusCode::UNAUTHORIZED);

    let authorized = authorized_client(&server).await;
    let foreign_endpoint = endpoint.replace("ws_default", foreign_workspace.as_str());
    let forbidden = authorized
        .post(foreign_endpoint)
        .header("content-type", "multipart/form-data; boundary=broken")
        .body("this body must not be parsed before workspace authorization")
        .send()
        .await
        .unwrap();
    assert_eq!(forbidden.status(), reqwest::StatusCode::FORBIDDEN);
    assert_eq!(
        sqlx::query_scalar::<_, i64>("select count(*) from artifact_sources")
            .fetch_one(registry.pool())
            .await
            .unwrap(),
        0
    );
    assert_eq!(
        sqlx::query_scalar::<_, i64>("select count(*) from import_jobs")
            .fetch_one(registry.pool())
            .await
            .unwrap(),
        0
    );
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn parser_canary_never_reaches_diagnostics_logs_traces_or_metrics() {
    const CANARY: &str = "openapi-telemetry-secret-canary";

    let recorder = DebuggingRecorder::new();
    let snapshotter = recorder.snapshotter();
    recorder
        .install()
        .expect("isolated integration test metrics recorder");

    let writer = SharedLogWriter::default();
    let exported = Arc::new(Mutex::new(Vec::new()));
    let provider = SdkTracerProvider::builder()
        .with_simple_exporter(CapturingExporter(Arc::clone(&exported)))
        .build();
    let tracer = provider.tracer("admin-openapi-source-test");
    let subscriber = tracing_subscriber::registry()
        .with(tracing_subscriber::fmt::layer().with_writer(writer.clone()))
        .with(tracing_opentelemetry::layer().with_tracer(tracer));
    let dispatch = tracing::Dispatch::new(subscriber);
    tracing::dispatcher::set_global_default(dispatch)
        .expect("isolated integration test tracing subscriber");

    let registry = test_registry().await;
    let app = build_test_app(registry, test_storage_root("openapi_telemetry_canary"));
    let server = spawn_admin_api(app).await;
    let client = authorized_client(&server).await;
    let document = format!(
        "openapi: 3.0.3\ninfo: {{ title: Canary }}\npaths:\n  /broken:\n    get:\n      description: {CANARY}\n      responses: ["
    );
    let response = client
        .post(format!("{server}/imports/openapi/preview"))
        .multipart(Form::new().part(
            "file",
            file_part(document.as_bytes(), "openapi.yaml", "application/yaml"),
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST);
    let trace_id = response
        .headers()
        .get("x-trace-id")
        .unwrap()
        .to_str()
        .unwrap()
        .to_owned();
    let body = response.text().await.unwrap();
    assert!(!body.contains(CANARY));
    let diagnostics: Value = serde_json::from_str(&body).unwrap();
    assert_eq!(diagnostics["error"]["trace_id"], trace_id);

    provider.force_flush().unwrap();
    let logs = writer.output();
    assert!(!logs.contains(CANARY));
    assert!(logs.contains(&trace_id));
    let spans = exported.lock().unwrap();
    let rendered_spans = format!("{spans:?}");
    assert!(!rendered_spans.contains(CANARY));
    drop(spans);
    for (key, _, _, _) in snapshotter.snapshot().into_vec() {
        assert!(!key.key().name().contains(CANARY));
        assert!(
            !key.key()
                .labels()
                .any(|label| { label.key().contains(CANARY) || label.value().contains(CANARY) })
        );
    }
    provider.shutdown().unwrap();
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn multipart_boundary_rejections_are_localized_and_leave_no_entities() {
    let registry = test_registry().await;
    let app = build_test_app(
        registry.clone(),
        test_storage_root("openapi_multipart_boundary"),
    );
    let server = spawn_admin_api(app).await;
    let client = authorized_client(&server).await;

    assert_rejected(
        &client,
        &server,
        Form::new().part(
            "file",
            file_part(b"sensitive-openapi-body", "openapi.txt", "text/plain"),
        ),
        "ru-RU",
        reqwest::StatusCode::BAD_REQUEST,
        "openapi_upload.invalid_media_type",
        Some("тип"),
    )
    .await;
    assert_rejected(
        &client,
        &server,
        Form::new().part(
            "file",
            file_part(OPENAPI.as_bytes(), "openapi.txt", "application/yaml"),
        ),
        "en-US",
        reqwest::StatusCode::BAD_REQUEST,
        "openapi_upload.invalid_media_type",
        Some("not supported"),
    )
    .await;
    assert_rejected(
        &client,
        &server,
        Form::new().part(
            "file",
            file_part(OPENAPI.as_bytes(), "openapi.yaml", "application/json"),
        ),
        "en-US",
        reqwest::StatusCode::BAD_REQUEST,
        "openapi_upload.invalid_media_type",
        Some("not supported"),
    )
    .await;
    assert_rejected(
        &client,
        &server,
        Form::new(),
        "en-US",
        reqwest::StatusCode::BAD_REQUEST,
        "openapi_upload.malformed_multipart",
        Some("multipart"),
    )
    .await;
    assert_eq!(
        sqlx::query_scalar::<_, i64>("select count(*) from import_jobs")
            .fetch_one(registry.pool())
            .await
            .unwrap(),
        0
    );
    assert_eq!(
        sqlx::query_scalar::<_, i64>("select count(*) from operations")
            .fetch_one(registry.pool())
            .await
            .unwrap(),
        0
    );
    assert_rejected(
        &client,
        &server,
        Form::new().part(
            "other",
            file_part(OPENAPI.as_bytes(), "openapi.yaml", "application/yaml"),
        ),
        "en-US",
        reqwest::StatusCode::BAD_REQUEST,
        "openapi_upload.malformed_multipart",
        Some("multipart"),
    )
    .await;
    assert_rejected(
        &client,
        &server,
        Form::new()
            .part(
                "file",
                file_part(OPENAPI.as_bytes(), "one.yaml", "application/yaml"),
            )
            .part(
                "file",
                file_part(OPENAPI.as_bytes(), "two.yaml", "application/yaml"),
            ),
        "en-US",
        reqwest::StatusCode::BAD_REQUEST,
        "openapi_upload.malformed_multipart",
        Some("multipart"),
    )
    .await;
    assert_rejected(
        &client,
        &server,
        Form::new().part("file", file_part(b"", "openapi.yaml", "application/yaml")),
        "en-US",
        reqwest::StatusCode::BAD_REQUEST,
        "openapi_upload.empty_file",
        Some("empty"),
    )
    .await;
    assert_rejected(
        &client,
        &server,
        Form::new().part(
            "file",
            file_part([0xff, 0xfe], "openapi.yaml", "application/yaml"),
        ),
        "en-US",
        reqwest::StatusCode::BAD_REQUEST,
        "openapi_upload.invalid_utf8",
        Some("UTF-8"),
    )
    .await;
    assert_rejected(
        &client,
        &server,
        Form::new().part(
            "file",
            file_part(
                vec![b'x'; MAX_ARTIFACT_BYTES + 1],
                "openapi.yaml",
                "application/yaml",
            ),
        ),
        "en-US",
        reqwest::StatusCode::PAYLOAD_TOO_LARGE,
        "openapi_upload.file_too_large",
        Some("256 KiB"),
    )
    .await;
    assert_eq!(
        sqlx::query_scalar::<_, i64>("select count(*) from import_jobs")
            .fetch_one(registry.pool())
            .await
            .unwrap(),
        0
    );
    assert_eq!(
        sqlx::query_scalar::<_, i64>("select count(*) from artifact_sources")
            .fetch_one(registry.pool())
            .await
            .unwrap(),
        0
    );

    assert_rejected(
        &client,
        &server,
        Form::new().part(
            "file",
            file_part(
                b"openapi: 3.0.3\ninfo: { title: Empty }\npaths: {}\n",
                "openapi.yaml",
                "application/yaml",
            ),
        ),
        "en-US",
        reqwest::StatusCode::BAD_REQUEST,
        "openapi_upload.no_methods",
        Some("no supported methods"),
    )
    .await;
    assert_rejected(
        &client,
        &server,
        Form::new().part("file", file_part(b"", "openapi.yaml", "application/yaml")),
        "ru-RU",
        reqwest::StatusCode::BAD_REQUEST,
        "openapi_upload.empty_file",
        Some("пуст"),
    )
    .await;
    assert_rejected(
        &client,
        &server,
        Form::new().part(
            "file",
            file_part(b"openapi: [", "openapi.yaml", "application/yaml"),
        ),
        "ru-RU",
        reqwest::StatusCode::BAD_REQUEST,
        "openapi_upload.invalid_document",
        Some("документ"),
    )
    .await;
    wait_for_source_lifecycle(&registry, ArtifactSourceLifecycle::Detached).await;

    let mut exact_limit = OPENAPI.as_bytes().to_vec();
    let padding = MAX_ARTIFACT_BYTES
        .checked_sub(exact_limit.len())
        .expect("OpenAPI fixture must fit inside the exact-size boundary");
    exact_limit.extend(std::iter::repeat_n(b'#', padding));
    let response = client
        .post(format!("{server}/imports/openapi/preview"))
        .multipart(Form::new().part(
            "file",
            file_part(&exact_limit, "openapi.yaml", "application/yaml"),
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), reqwest::StatusCode::OK);
    let body = response.json::<Value>().await.unwrap();
    let rendered = body.to_string();
    assert!(body["job_id"].is_string());
    assert!(!rendered.contains("source_id"));
    assert!(!rendered.contains("digest"));
}

async fn wait_for_source_lifecycle(
    registry: &crank_registry::PostgresRegistry,
    lifecycle: ArtifactSourceLifecycle,
) {
    let expected = match lifecycle {
        ArtifactSourceLifecycle::Active => "active",
        ArtifactSourceLifecycle::Detached => "detached",
    };
    for _ in 0..100 {
        let found = sqlx::query_scalar::<_, bool>(
            "select exists(select 1 from artifact_sources where lifecycle = $1)",
        )
        .bind(expected)
        .fetch_one(registry.pool())
        .await
        .unwrap();
        if found {
            return;
        }
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }
    panic!("artifact source did not reach {expected}");
}

fn file_part(bytes: impl AsRef<[u8]>, filename: &str, mime_type: &str) -> Part {
    Part::bytes(bytes.as_ref().to_vec())
        .file_name(filename.to_owned())
        .mime_str(mime_type)
        .unwrap()
}

async fn assert_rejected(
    client: &reqwest::Client,
    server: &impl AsRef<str>,
    form: Form,
    language: &str,
    expected_status: reqwest::StatusCode,
    expected_code: &str,
    message_fragment: Option<&str>,
) {
    let response = client
        .post(format!("{}/imports/openapi/preview", server.as_ref()))
        .header("accept-language", language)
        .multipart(form)
        .send()
        .await
        .unwrap();
    let status = response.status();
    let body = response.json::<Value>().await.unwrap();
    assert_eq!(
        status, expected_status,
        "unexpected status while checking {expected_code}: {body}"
    );
    assert_eq!(body["error"]["code"], expected_code);
    assert!(body["error"]["request_id"].is_string());
    assert!(body["error"]["trace_id"].is_string());
    if let Some(fragment) = message_fragment {
        assert!(
            body["error"]["message"]
                .as_str()
                .unwrap()
                .contains(fragment)
        );
    }
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn apply_rereads_the_scoped_verified_source_and_expiry_detaches_it() {
    let registry = test_registry().await;
    let service = test_service(
        registry.clone(),
        test_storage_root("openapi_verified_source"),
        test_auth_settings(),
        test_secret_crypto(),
    );
    let service = service.clone();
    let workspace_id = WorkspaceId::new("ws_default");
    let preview = service
        .preview_openapi_import(
            &workspace_id,
            OpenApiUpload {
                bytes: OPENAPI.as_bytes().to_vec(),
                mime_type: "application/yaml".to_owned(),
                locale: OpenApiUploadLocale::En,
            },
        )
        .await
        .unwrap();
    let job_id = preview.job_id.as_str().into();
    let job = registry
        .get_import_job(&workspace_id, &job_id)
        .await
        .unwrap()
        .unwrap();
    let source_id = job.preview_payload["source"]["source_id"]
        .as_str()
        .unwrap()
        .to_owned();
    let source = registry
        .get_artifact_source(&workspace_id, &source_id.as_str().into())
        .await
        .unwrap();
    assert_eq!(source.lifecycle, ArtifactSourceLifecycle::Active);

    // The stored preview is presentation data only. A forged candidate cannot
    // create a Draft because apply reparses the source bytes.
    sqlx::query("update import_jobs set preview_payload = jsonb_set(preview_payload, '{preview}', '{\"groups\":[]}') where id = $1")
        .bind(job_id.as_str())
        .execute(registry.pool())
        .await
        .unwrap();
    let imported = service
        .create_openapi_import(
            &workspace_id,
            &job_id,
            OpenApiImportCreatePayload {
                selected_operation_keys: vec!["GET /health".to_owned()],
                server_url: None,
                conflict_mode: "skip".to_owned(),
            },
        )
        .await
        .unwrap();
    assert_eq!(imported.created.len(), 1);
    let detached = registry
        .get_artifact_source(&workspace_id, &source_id.as_str().into())
        .await
        .unwrap();
    assert_eq!(detached.lifecycle, ArtifactSourceLifecycle::Detached);

    let second_preview = service
        .preview_openapi_import(
            &workspace_id,
            OpenApiUpload {
                bytes: OPENAPI.as_bytes().to_vec(),
                mime_type: "application/yaml".to_owned(),
                locale: OpenApiUploadLocale::En,
            },
        )
        .await
        .unwrap();
    let second_job_id = second_preview.job_id.as_str().into();
    let second_job = registry
        .get_import_job(&workspace_id, &second_job_id)
        .await
        .unwrap()
        .unwrap();
    let second_source_id = second_job.preview_payload["source"]["source_id"]
        .as_str()
        .unwrap()
        .to_owned();
    sqlx::query("update import_jobs set expires_at = $1 where id = $2")
        .bind(OffsetDateTime::now_utc() - Duration::minutes(1))
        .bind(second_job_id.as_str())
        .execute(registry.pool())
        .await
        .unwrap();
    registry.delete_expired_import_jobs().await.unwrap();
    let expired_job_source = registry
        .get_artifact_source(&workspace_id, &second_source_id.as_str().into())
        .await
        .unwrap();
    assert_eq!(
        expired_job_source.lifecycle,
        ArtifactSourceLifecycle::Detached
    );

    let legacy_preview = service
        .preview_openapi_import(
            &workspace_id,
            OpenApiUpload {
                bytes: format!("{OPENAPI}\n# legacy-upgrade").into_bytes(),
                mime_type: "application/yaml".to_owned(),
                locale: OpenApiUploadLocale::En,
            },
        )
        .await
        .unwrap();
    let legacy_job_id: crank_registry::ImportJobId = legacy_preview.job_id.as_str().into();
    let legacy_job = registry
        .get_import_job(&workspace_id, &legacy_job_id)
        .await
        .unwrap()
        .unwrap();
    let legacy_source_id = legacy_job.preview_payload["source"]["source_id"]
        .as_str()
        .unwrap()
        .to_owned();
    sqlx::query(
        "update import_jobs
            set preview_payload = jsonb_set(
                    preview_payload,
                    '{source}',
                    '{\"format\":\"openapi\",\"version\":\"3.0.3\"}'::jsonb
                ),
                expires_at = now() - interval '1 minute'
          where id = $1",
    )
    .bind(legacy_job_id.as_str())
    .execute(registry.pool())
    .await
    .unwrap();
    sqlx::query(
        "update artifact_sources
            set created_at = now() - interval '10 minutes',
                updated_at = now() - interval '10 minutes'
          where workspace_id = $1 and source_id = $2",
    )
    .bind(workspace_id.as_str())
    .bind(&legacy_source_id)
    .execute(registry.pool())
    .await
    .unwrap();

    registry.delete_expired_import_jobs().await.unwrap();
    assert!(
        registry
            .get_import_job(&workspace_id, &legacy_job_id)
            .await
            .unwrap()
            .is_none()
    );
    assert_eq!(
        registry
            .get_artifact_source(&workspace_id, &legacy_source_id.as_str().into())
            .await
            .unwrap()
            .lifecycle,
        ArtifactSourceLifecycle::Detached
    );
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn cancellation_detaches_and_restart_cleanup_recovers_a_dangling_source() {
    let registry = test_registry().await;
    let service = test_service(
        registry.clone(),
        test_storage_root("openapi_cancel_restart"),
        test_auth_settings(),
        test_secret_crypto(),
    );
    let schema = sqlx::query_scalar::<_, String>("select current_schema()")
        .fetch_one(registry.pool())
        .await
        .unwrap();
    let observer = sqlx::postgres::PgPoolOptions::new()
        .max_connections(1)
        .connect(crank_test_support::postgres_database_url().await)
        .await
        .unwrap();
    let lock_pool = sqlx::postgres::PgPoolOptions::new()
        .max_connections(1)
        .connect(crank_test_support::postgres_database_url().await)
        .await
        .unwrap();
    sqlx::query("select set_config('search_path', $1, false)")
        .bind(&schema)
        .execute(&observer)
        .await
        .unwrap();
    sqlx::query("select set_config('search_path', $1, false)")
        .bind(&schema)
        .execute(&lock_pool)
        .await
        .unwrap();
    sqlx::query(
        "create function block_openapi_import_insert() returns trigger language plpgsql as $$
         begin
           perform pg_advisory_xact_lock(2147483001);
           return new;
         end $$",
    )
    .execute(registry.pool())
    .await
    .unwrap();
    sqlx::query(
        "create trigger block_openapi_import_insert
         before insert on import_jobs
         for each row execute function block_openapi_import_insert()",
    )
    .execute(registry.pool())
    .await
    .unwrap();
    let mut lock_connection = lock_pool.acquire().await.unwrap();
    sqlx::query("select pg_advisory_lock(2147483001)")
        .execute(&mut *lock_connection)
        .await
        .unwrap();

    let workspace_id = WorkspaceId::new("ws_default");
    let preview_task = tokio::spawn({
        let service = service.clone();
        let workspace_id = workspace_id.clone();
        async move {
            service
                .preview_openapi_import(
                    &workspace_id,
                    OpenApiUpload {
                        bytes: OPENAPI.as_bytes().to_vec(),
                        mime_type: "application/yaml".to_owned(),
                        locale: OpenApiUploadLocale::En,
                    },
                )
                .await
        }
    });
    let cancelled_source_id =
        wait_for_source_lifecycle_in_pool(&observer, ArtifactSourceLifecycle::Active).await;
    assert!(!preview_task.is_finished());
    preview_task.abort();
    let cancellation = preview_task.await.unwrap_err();
    assert!(cancellation.is_cancelled());
    wait_for_blocked_import_insert_to_stop(&observer).await;
    sqlx::query("select pg_advisory_unlock(2147483001)")
        .execute(&mut *lock_connection)
        .await
        .unwrap();
    wait_for_specific_source_lifecycle(
        &registry,
        &cancelled_source_id,
        ArtifactSourceLifecycle::Detached,
    )
    .await;
    assert_eq!(
        sqlx::query_scalar::<_, i64>("select count(*) from import_jobs")
            .fetch_one(registry.pool())
            .await
            .unwrap(),
        0
    );

    sqlx::query("drop trigger block_openapi_import_insert on import_jobs")
        .execute(registry.pool())
        .await
        .unwrap();
    let preview = service
        .preview_openapi_import(
            &workspace_id,
            OpenApiUpload {
                bytes: format!("{OPENAPI}\n# restart-window").into_bytes(),
                mime_type: "application/yaml".to_owned(),
                locale: OpenApiUploadLocale::En,
            },
        )
        .await
        .unwrap();
    let job_id: crank_registry::ImportJobId = preview.job_id.as_str().into();
    let job = registry
        .get_import_job(&workspace_id, &job_id)
        .await
        .unwrap()
        .unwrap();
    let source_id = job.preview_payload["source"]["source_id"]
        .as_str()
        .unwrap()
        .to_owned();
    sqlx::query("delete from import_jobs where id = $1")
        .bind(job_id.as_str())
        .execute(registry.pool())
        .await
        .unwrap();
    sqlx::query(
        "update artifact_sources
            set created_at = now() - interval '10 minutes',
                updated_at = now() - interval '10 minutes'
          where workspace_id = $1 and source_id = $2",
    )
    .bind(workspace_id.as_str())
    .bind(&source_id)
    .execute(registry.pool())
    .await
    .unwrap();

    registry.delete_expired_import_jobs().await.unwrap();
    let recovered = registry
        .get_artifact_source(&workspace_id, &source_id.as_str().into())
        .await
        .unwrap();
    assert_eq!(recovered.lifecycle, ArtifactSourceLifecycle::Detached);
}

async fn wait_for_source_lifecycle_in_pool(
    pool: &sqlx::PgPool,
    lifecycle: ArtifactSourceLifecycle,
) -> ArtifactSourceId {
    let expected = match lifecycle {
        ArtifactSourceLifecycle::Active => "active",
        ArtifactSourceLifecycle::Detached => "detached",
    };
    for _ in 0..100 {
        let found = sqlx::query_scalar::<_, Option<String>>(
            "select min(source_id) from artifact_sources where lifecycle = $1",
        )
        .bind(expected)
        .fetch_one(pool)
        .await
        .unwrap();
        if let Some(source_id) = found {
            return ArtifactSourceId::new(source_id);
        }
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }
    panic!("artifact source did not reach {expected}");
}

async fn wait_for_blocked_import_insert_to_stop(pool: &sqlx::PgPool) {
    for _ in 0..100 {
        let active = sqlx::query_scalar::<_, bool>(
            "select exists(
                select 1
                  from pg_stat_activity
                 where state = 'active'
                   and query like 'insert into import_jobs%'
            )",
        )
        .fetch_one(pool)
        .await
        .unwrap();
        if !active {
            return;
        }
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }
    panic!("cancelled import insert remained active in PostgreSQL");
}

async fn wait_for_specific_source_lifecycle(
    registry: &crank_registry::PostgresRegistry,
    source_id: &ArtifactSourceId,
    lifecycle: ArtifactSourceLifecycle,
) {
    let expected = match lifecycle {
        ArtifactSourceLifecycle::Active => "active",
        ArtifactSourceLifecycle::Detached => "detached",
    };
    for _ in 0..100 {
        let found = sqlx::query_scalar::<_, bool>(
            "select exists(
                select 1
                  from artifact_sources
                 where source_id = $1 and lifecycle = $2
            )",
        )
        .bind(source_id.as_str())
        .bind(expected)
        .fetch_one(registry.pool())
        .await
        .unwrap();
        if found {
            return;
        }
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }
    panic!("artifact source {source_id:?} did not reach {expected}");
}

#[derive(Clone, Default)]
struct SharedLogWriter {
    buffer: Arc<Mutex<Vec<u8>>>,
}

impl SharedLogWriter {
    fn output(&self) -> String {
        String::from_utf8(self.buffer.lock().unwrap().clone()).unwrap()
    }
}

impl<'a> MakeWriter<'a> for SharedLogWriter {
    type Writer = SharedLogGuard;

    fn make_writer(&'a self) -> Self::Writer {
        SharedLogGuard {
            buffer: Arc::clone(&self.buffer),
        }
    }
}

struct SharedLogGuard {
    buffer: Arc<Mutex<Vec<u8>>>,
}

impl io::Write for SharedLogGuard {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        self.buffer.lock().unwrap().extend_from_slice(bytes);
        Ok(bytes.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

#[derive(Clone, Debug)]
struct CapturingExporter(Arc<Mutex<Vec<SpanData>>>);

impl SpanExporter for CapturingExporter {
    async fn export(&self, batch: Vec<SpanData>) -> OTelSdkResult {
        self.0.lock().unwrap().extend(batch);
        Ok(())
    }
}