use std::net::SocketAddr;
use std::sync::{Arc, Barrier, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use remora_admission::{AdmissionConfig, AdmissionDecision, AdmissionService};
use remora_attempt::{
    ActivationOutcome, AttemptConfig, AttemptService, CancelOutcome, CleanupCompletion,
    CleanupIdentity, CooperativeStopOutcome, RecoveryConfig, StagingMultipart, StagingObject,
    StagingWriteIntent,
};
use remora_contracts::{HttpSourceSubmissionDto, JobSubmissionDto, SourceSubmissionDto};
use remora_coordination::{CoordinationStore, StoreConfig};
use remora_domain::{
    BucketName, OpaqueId, RuntimeFailureClass, StorageScope, TargetPrefix, TargetStoreId,
};
use remora_durable::{PoolConfig, connect};
use remora_identity::{DefaultTarget, Principal};
use remora_result::PublicationInput;
use remora_schema::migrate as migrate_schema;
use remora_target::{
    CreateObjectRequest, MultipartIdentity, ObjectIdentity, StoredObject, TargetStore,
    TargetStoreConfig, attempt_manifest_key_in, attempt_object_key_in,
};
use sqlx::PgPool;
use tokio::io::copy_bidirectional;
use tokio::net::{TcpListener, TcpStream};
use tokio::task::JoinHandle;

struct Harness {
    pool: PgPool,
    coordination: CoordinationStore,
    target: TargetStore,
    target_config: TargetStoreConfig,
}

struct CoordinationProxy {
    address: SocketAddr,
    accept_task: JoinHandle<()>,
    connections: Arc<Mutex<Vec<JoinHandle<()>>>>,
}

struct GateFault {
    point: remora_attempt::AttemptFaultPoint,
    entered: Barrier,
    release: Barrier,
}

impl GateFault {
    fn new(point: remora_attempt::AttemptFaultPoint) -> Self {
        Self {
            point,
            entered: Barrier::new(2),
            release: Barrier::new(2),
        }
    }
}

impl remora_attempt::AttemptFaultInjector for GateFault {
    fn hit(&self, point: remora_attempt::AttemptFaultPoint) {
        if point == self.point {
            self.entered.wait();
            self.release.wait();
        }
    }
}

impl CoordinationProxy {
    async fn start(upstream: String, address: Option<SocketAddr>) -> Self {
        let listener = TcpListener::bind(address.unwrap_or_else(|| "127.0.0.1:0".parse().unwrap()))
            .await
            .unwrap();
        let address = listener.local_addr().unwrap();
        let connections = Arc::new(Mutex::new(Vec::new()));
        let active = connections.clone();
        let accept_task = tokio::spawn(async move {
            while let Ok((mut downstream, _)) = listener.accept().await {
                let Ok(mut upstream_stream) = TcpStream::connect(&upstream).await else {
                    continue;
                };
                let task = tokio::spawn(async move {
                    let _ = copy_bidirectional(&mut downstream, &mut upstream_stream).await;
                });
                active.lock().unwrap().push(task);
            }
        });
        Self {
            address,
            accept_task,
            connections,
        }
    }

    async fn partition(self) -> SocketAddr {
        self.accept_task.abort();
        let _ = self.accept_task.await;
        for task in self.connections.lock().unwrap().drain(..) {
            task.abort();
        }
        tokio::task::yield_now().await;
        self.address
    }

    fn url(&self) -> String {
        format!("redis://{}/", self.address)
    }
}

impl Harness {
    async fn new(name: &str) -> Option<Self> {
        let database_url = required_env("DATABASE_URL")?;
        let coordination_url = required_env("REMORA_COORDINATION_URL")?;
        let target_config = match TargetStoreConfig::from_env() {
            Ok(config) => config,
            Err(error) if required() => panic!("target fixture is required: {error}"),
            Err(_) => return None,
        };
        let target = TargetStore::connect(target_config.clone()).expect("connect target");
        let pool = connect(
            &database_url,
            PoolConfig {
                max_connections: 16,
                acquire_timeout: Duration::from_secs(10),
                idle_timeout: Some(Duration::from_secs(30)),
            },
        )
        .await
        .expect("connect postgres");
        migrate_schema(&pool).await.expect("schema migrate");
        sqlx::query("TRUNCATE TABLE jobs CASCADE")
            .execute(&pool)
            .await
            .expect("reset recovery test state");
        let coordination = CoordinationStore::connect(
            &coordination_url,
            format!("{}wp17:{name}:{}:", coordination_namespace(), nonce()),
            StoreConfig {
                global_limit: 32,
                lease_ttl: Duration::from_secs(10),
                reserve_ttl: Duration::from_secs(10),
                progress_ttl: Duration::from_secs(10),
                cancel_ttl: None,
            },
        )
        .await
        .expect("connect coordination store");
        Some(Self {
            pool,
            coordination,
            target,
            target_config,
        })
    }

    fn service(&self, max_attempts: u64) -> AttemptService {
        AttemptService::new(
            self.pool.clone(),
            self.coordination.clone(),
            AttemptConfig { max_attempts },
        )
        .expect("attempt service")
    }

    async fn job(&self) -> OpaqueId {
        let store = TargetStoreId::parse(&self.target_config.target_store_id).unwrap();
        let bucket = BucketName::parse(&self.target_config.bucket).unwrap();
        let prefix = TargetPrefix::parse("users/recovery", ["_remora"]).unwrap();
        let owner = unique_id("owner");
        let principal = Principal {
            id: owner,
            issuer: "https://issuer.example".into(),
            subject: unique_id("subject").to_string(),
            administrator: false,
            scopes: [StorageScope {
                store: store.clone(),
                bucket: bucket.clone(),
                prefix: prefix.clone(),
            }]
            .into_iter()
            .collect(),
        };
        let admission = AdmissionService::new(
            self.pool.clone(),
            AdmissionConfig {
                default_target: DefaultTarget {
                    store,
                    bucket,
                    prefix,
                },
                reserved_target_segments: vec!["_remora".into()],
                policy_version: "recovery-tests-v1".into(),
            },
        )
        .unwrap();
        let outcome = admission
            .submit(
                &principal,
                unique_id("submit").as_str(),
                JobSubmissionDto {
                    source: SourceSubmissionDto::Http(HttpSourceSubmissionDto {
                        url: "https://example.invalid/recovery".into(),
                        result_path: "artifact.bin".into(),
                        credential_id: None,
                    }),
                    target: None,
                    label: Some("recovery-test".into()),
                },
            )
            .await
            .expect("submit job");
        let AdmissionDecision::Accepted(job) = outcome.decision else {
            panic!("job rejected")
        };
        job.job_id
    }

    async fn executing(
        &self,
        service: &AttemptService,
        job: &OpaqueId,
    ) -> remora_attempt::AttemptAuthority {
        self.coordination
            .enqueue(job.as_str())
            .await
            .expect("enqueue");
        let claim = service
            .claim(job, &unique_id("worker"), &unique_id("claimop"))
            .await
            .expect("claim");
        let remora_attempt::ClaimOutcome::Claimed { authority, .. } = claim else {
            panic!("claim rejected")
        };
        assert!(matches!(
            service
                .activate(&authority, &unique_id("activateop"))
                .await
                .expect("activate"),
            ActivationOutcome::Activated(_)
        ));
        service
            .authority(&authority.attempt_id)
            .await
            .unwrap()
            .unwrap()
    }

    async fn seal_result(
        &self,
        authority: &remora_attempt::AttemptAuthority,
        manifest: StoredObject,
    ) -> PublicationInput {
        let manifest_id = unique_id("manifest");
        sqlx::query(
            "INSERT INTO result_inventories (
                attempt_id, state, manifest_schema_version, manifest_id, manifest_store,
                manifest_bucket, manifest_key, manifest_version_id, manifest_sha256,
                manifest_size, member_count, total_bytes,
                staging_origin_attempt_id, target_prefix
             ) VALUES (
                $1, 'sealed', 1, $2, $3, $4, $5, $6, $7, $8, 0, 0,
                $1, 'users/recovery'
             )",
        )
        .bind(authority.attempt_id.as_str())
        .bind(manifest_id.as_str())
        .bind(&manifest.identity.target_store_id)
        .bind(&manifest.identity.bucket)
        .bind(&manifest.identity.key)
        .bind(&manifest.identity.version_id)
        .bind(manifest.content_sha256.as_str())
        .bind(i64::try_from(manifest.size).unwrap())
        .execute(&self.pool)
        .await
        .unwrap();
        PublicationInput {
            attempt_id: authority.attempt_id.clone(),
            manifest_id,
            schema_version: 1,
            manifest,
            member_count: 0,
            total_bytes: 0,
        }
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn activation_and_recovery_share_authority_serialization() {
    let Some(harness) = Harness::new("activation-race").await else {
        return;
    };
    let activation_gate = Arc::new(GateFault::new(
        remora_attempt::AttemptFaultPoint::ActivationAfterLease,
    ));
    let service = AttemptService::new_with_fault_injector(
        harness.pool.clone(),
        harness.coordination.clone(),
        AttemptConfig { max_attempts: 3 },
        activation_gate.clone(),
    )
    .unwrap();
    let activation_job = harness.job().await;
    harness
        .coordination
        .enqueue(activation_job.as_str())
        .await
        .unwrap();
    let activation_claim = service
        .claim(&activation_job, &unique_id("worker"), &unique_id("claimop"))
        .await
        .unwrap();
    let remora_attempt::ClaimOutcome::Claimed {
        authority: activation_authority,
        ..
    } = activation_claim
    else {
        panic!("claim rejected")
    };
    let activating = service.clone();
    let activation_input = activation_authority.clone();
    let activation = tokio::spawn(async move {
        activating
            .activate(&activation_input, &unique_id("activateop"))
            .await
            .unwrap()
    });
    activation_gate.entered.wait();
    let recovering = service.clone();
    let recovery =
        tokio::spawn(async move { recovering.reconcile_once(&recovery_config()).await.unwrap() });
    tokio::task::yield_now().await;
    activation_gate.release.wait();
    assert!(matches!(
        activation.await.unwrap(),
        ActivationOutcome::Activated(_)
    ));
    assert_eq!(recovery.await.unwrap().attempts_reaped, 0);
    assert_eq!(
        service
            .authority(&activation_authority.attempt_id)
            .await
            .unwrap()
            .unwrap()
            .state,
        remora_attempt::AttemptState::Executing
    );
    service
        .publish_failure(
            &activation_authority,
            &unique_id("failure"),
            RuntimeFailureClass::Authorization,
        )
        .await
        .unwrap();

    let recovery_gate = Arc::new(GateFault::new(
        remora_attempt::AttemptFaultPoint::RecoveryBeforeRevalidation,
    ));
    let service = AttemptService::new_with_fault_injector(
        harness.pool.clone(),
        harness.coordination.clone(),
        AttemptConfig { max_attempts: 3 },
        recovery_gate.clone(),
    )
    .unwrap();
    let recovery_job = harness.job().await;
    harness
        .coordination
        .enqueue(recovery_job.as_str())
        .await
        .unwrap();
    let worker = unique_id("worker");
    let recovery_claim = service
        .claim(&recovery_job, &worker, &unique_id("claimop"))
        .await
        .unwrap();
    let remora_attempt::ClaimOutcome::Claimed {
        authority: recovery_authority,
        ..
    } = recovery_claim
    else {
        panic!("claim rejected")
    };
    assert_eq!(
        service.cooperative_stop(&recovery_authority).await.unwrap(),
        CooperativeStopOutcome::StopNotRequired
    );
    harness
        .coordination
        .free_reserve(recovery_job.as_str(), Some(worker.as_str()))
        .await
        .unwrap();
    let recovering = service.clone();
    let recovery =
        tokio::spawn(async move { recovering.reconcile_once(&recovery_config()).await.unwrap() });
    recovery_gate.entered.wait();
    let activating = service.clone();
    let activation_input = recovery_authority.clone();
    let activation = tokio::spawn(async move {
        activating
            .activate(&activation_input, &unique_id("activateop"))
            .await
            .unwrap()
    });
    tokio::task::yield_now().await;
    recovery_gate.release.wait();
    assert_eq!(recovery.await.unwrap().attempts_reaped, 1);
    assert_eq!(
        activation.await.unwrap(),
        ActivationOutcome::PreconditionFailed
    );
    assert!(
        service
            .authority(&recovery_authority.attempt_id)
            .await
            .unwrap()
            .is_none()
    );
}

#[tokio::test]
async fn batched_reconciliation_rotates_across_eligible_jobs() {
    let Some(harness) = Harness::new("fairness").await else {
        return;
    };
    let service = harness.service(3);
    let first = harness.job().await;
    let second = harness.job().await;
    sqlx::query(
        "UPDATE jobs SET recovery_checked_at = $1
         WHERE job_id <> $2 AND job_id <> $3",
    )
    .bind(time::OffsetDateTime::now_utc())
    .bind(first.as_str())
    .bind(second.as_str())
    .execute(&harness.pool)
    .await
    .unwrap();
    let config = RecoveryConfig {
        batch_size: 1,
        ..recovery_config()
    };
    assert_eq!(service.reconcile_once(&config).await.unwrap().enqueued, 1);
    assert_eq!(service.reconcile_once(&config).await.unwrap().enqueued, 1);
    let mut projected = [
        harness.coordination.dequeue().await.unwrap().unwrap(),
        harness.coordination.dequeue().await.unwrap().unwrap(),
    ];
    projected.sort();
    let mut expected = [first.to_string(), second.to_string()];
    expected.sort();
    assert_eq!(projected, expected);
}

#[tokio::test]
async fn coordination_partition_rolls_back_and_restart_converges() {
    let Some(harness) = Harness::new("partition").await else {
        return;
    };
    let upstream = coordination_upstream();
    let proxy = CoordinationProxy::start(upstream.clone(), None).await;
    let proxy_url = proxy.url();
    let store = CoordinationStore::connect(
        &proxy_url,
        format!("{}wp17:partition:{}:", coordination_namespace(), nonce()),
        StoreConfig::default(),
    )
    .await
    .unwrap();
    let service = AttemptService::new(
        harness.pool.clone(),
        store.clone(),
        AttemptConfig { max_attempts: 3 },
    )
    .unwrap();
    let job = harness.job().await;
    assert_eq!(
        service
            .reconcile_once(&recovery_config())
            .await
            .unwrap()
            .enqueued,
        1
    );
    store.drop_job_state(&[job.as_str()]).await.unwrap();
    let address = proxy.partition().await;
    let interrupted = tokio::time::timeout(
        Duration::from_millis(250),
        service.reconcile_once(&recovery_config()),
    )
    .await;
    assert!(interrupted.is_err() || interrupted.unwrap().is_err());
    assert_eq!(
        service
            .mutation_projection(&job)
            .await
            .unwrap()
            .unwrap()
            .phase,
        remora_domain::JobPhase::Queued
    );

    let resumed_proxy = CoordinationProxy::start(upstream, Some(address)).await;
    let resumed_store = CoordinationStore::connect(
        &resumed_proxy.url(),
        format!("{}wp17:partition:{}:", coordination_namespace(), nonce()),
        StoreConfig::default(),
    )
    .await
    .unwrap();
    let restarted = AttemptService::new(
        harness.pool.clone(),
        resumed_store.clone(),
        AttemptConfig { max_attempts: 3 },
    )
    .unwrap();
    assert_eq!(
        restarted
            .reconcile_once(&recovery_config())
            .await
            .unwrap()
            .enqueued,
        1
    );
    assert_eq!(
        resumed_store.dequeue().await.unwrap(),
        Some(job.to_string())
    );
    resumed_proxy.partition().await;
}

#[tokio::test]
async fn coordination_wipes_reap_lost_attempts_and_reconstruct_dispatch() {
    let Some(harness) = Harness::new("wipe").await else {
        return;
    };
    let service = harness.service(3);
    let config = recovery_config();
    let queued = harness.job().await;
    let first = service.reconcile_once(&config).await.unwrap();
    assert_eq!(first.enqueued, 1);
    assert!(harness.coordination.dequeue().await.unwrap().is_some());
    harness
        .coordination
        .drop_job_state(&[queued.as_str()])
        .await
        .unwrap();
    let restarted = harness.service(3);
    let second = restarted.reconcile_once(&config).await.unwrap();
    assert_eq!(second.enqueued, 1);
    harness
        .coordination
        .drop_job_state(&[queued.as_str()])
        .await
        .unwrap();
    assert_eq!(restarted.reconcile_once(&config).await.unwrap().enqueued, 1);

    let authority = harness.executing(&service, &queued).await;
    service
        .report_progress(
            &authority,
            &unique_id("progress"),
            remora_attempt::ProgressRecord {
                completed_bytes: 64,
                total_bytes: Some(128),
                completed_members: 0,
                total_members: Some(1),
            },
        )
        .await
        .unwrap();
    harness
        .coordination
        .drop_job_state(&[queued.as_str()])
        .await
        .unwrap();
    assert_eq!(
        service
            .reconcile_once(&config)
            .await
            .unwrap()
            .attempts_reaped,
        1
    );
    let scheduled = service.reconcile_once(&config).await.unwrap();
    assert_eq!(scheduled.retries_scheduled, 1);
    assert_eq!(scheduled.enqueued, 1);
    assert!(
        harness
            .coordination
            .get_progress(queued.as_str())
            .await
            .unwrap()
            .is_none()
    );

    let worker = unique_id("worker");
    let claim = service
        .claim(&queued, &worker, &unique_id("claimop"))
        .await
        .unwrap();
    let remora_attempt::ClaimOutcome::Claimed { authority, .. } = claim else {
        panic!("retry claim rejected")
    };
    harness
        .coordination
        .free_reserve(queued.as_str(), Some(worker.as_str()))
        .await
        .unwrap();
    assert_eq!(
        service
            .reconcile_once(&config)
            .await
            .unwrap()
            .attempts_reaped,
        1
    );
    assert_eq!(
        sqlx::query_scalar::<_, String>("SELECT state FROM attempts WHERE attempt_id = $1")
            .bind(authority.attempt_id.as_str())
            .fetch_one(&harness.pool)
            .await
            .unwrap(),
        "reaped"
    );
}

#[tokio::test]
async fn cancellation_is_terminal_and_cooperative_stop_revokes_execution() {
    let Some(harness) = Harness::new("cancel").await else {
        return;
    };
    let service = harness.service(3);
    let config = recovery_config();
    let queued = harness.job().await;
    assert_eq!(
        service.request_cancel(&queued).await.unwrap(),
        CancelOutcome::Requested
    );
    assert_eq!(
        service.request_cancel(&queued).await.unwrap(),
        CancelOutcome::AlreadyTerminal
    );
    assert_eq!(
        service
            .mutation_projection(&queued)
            .await
            .unwrap()
            .unwrap()
            .phase,
        remora_domain::JobPhase::Cancelled
    );
    assert_eq!(
        service
            .reconcile_once(&config)
            .await
            .unwrap()
            .terminal_coordination_cleaned,
        1
    );

    let running = harness.job().await;
    let authority = harness.executing(&service, &running).await;
    assert_eq!(
        service.request_cancel(&running).await.unwrap(),
        CancelOutcome::Requested
    );
    let report = service.reconcile_once(&config).await.unwrap();
    assert_eq!(report.cancellations_published, 1);
    assert_eq!(report.cancellations_finalized, 1);
    assert_eq!(report.terminal_coordination_cleaned, 0);
    assert_eq!(harness.coordination.global_used().await.unwrap(), 1);
    assert_eq!(
        service.cooperative_stop(&authority).await.unwrap(),
        CooperativeStopOutcome::Stopped
    );
    assert_eq!(
        sqlx::query_scalar::<_, String>("SELECT state FROM attempts WHERE attempt_id = $1")
            .bind(authority.attempt_id.as_str())
            .fetch_one(&harness.pool)
            .await
            .unwrap(),
        "stopped"
    );
    assert_eq!(
        service
            .reconcile_once(&config)
            .await
            .unwrap()
            .terminal_coordination_cleaned,
        1
    );
    assert_eq!(harness.coordination.global_used().await.unwrap(), 0);
    assert_eq!(
        service
            .publish_failure(
                &authority,
                &unique_id("latefail"),
                RuntimeFailureClass::Internal,
            )
            .await
            .unwrap(),
        remora_attempt::AttemptMutationOutcome::Rejected {
            reason: remora_attempt::MutationRejection::AuthorityLost,
            disposition: remora_durable::MutationDisposition::First,
        }
    );
}

#[tokio::test]
async fn worker_loss_reaps_live_authority_without_treating_coordination_as_durable() {
    let Some(harness) = Harness::new("worker-down").await else {
        return;
    };
    let service = harness.service(1);
    let job = harness.job().await;
    let authority = harness.executing(&service, &job).await;
    assert_eq!(
        service
            .mark_worker_down(&authority.worker_incarnation)
            .await
            .unwrap(),
        1
    );
    let report = service.reconcile_once(&recovery_config()).await.unwrap();
    assert_eq!(report.attempts_reaped, 1);
    assert_eq!(
        service
            .mutation_projection(&job)
            .await
            .unwrap()
            .unwrap()
            .phase,
        remora_domain::JobPhase::Failed
    );
    assert_eq!(
        service.cooperative_stop(&authority).await.unwrap(),
        CooperativeStopOutcome::Stopped
    );
}

#[tokio::test]
async fn cleanup_uses_exact_durable_identities_retries_and_excludes_the_winner() {
    let Some(harness) = Harness::new("cleanup").await else {
        return;
    };
    let service = harness.service(3);

    let loser_job = harness.job().await;
    let loser = harness.executing(&service, &loser_job).await;
    let loser_key = attempt_object_key_in(
        "users/recovery",
        &loser.attempt_id,
        &unique_id("object"),
        None,
    );
    let loser_object = harness
        .target
        .put_create_only(
            &CreateObjectRequest::parse(loser_key.clone()).unwrap(),
            b"loser-stage".as_slice(),
        )
        .await
        .unwrap();
    service
        .register_staging_object(
            &loser.attempt_id,
            &StagingObject {
                target_store: loser_object.identity.target_store_id.clone(),
                bucket: loser_object.identity.bucket.clone(),
                key: loser_object.identity.key.clone(),
                version_id: loser_object.identity.version_id.clone(),
            },
        )
        .await
        .unwrap();
    assert!(
        service
            .register_staging_object(
                &loser.attempt_id,
                &StagingObject {
                    target_store: "foreign-store".into(),
                    bucket: loser_object.identity.bucket.clone(),
                    key: loser_object.identity.key.clone(),
                    version_id: loser_object.identity.version_id.clone(),
                },
            )
            .await
            .is_err()
    );
    let multipart_key = attempt_object_key_in(
        "users/recovery",
        &loser.attempt_id,
        &unique_id("multipart"),
        None,
    );
    let multipart = harness
        .target
        .begin_multipart(&CreateObjectRequest::parse(multipart_key.clone()).unwrap())
        .await
        .unwrap();
    let multipart_identity = multipart.identity().clone();
    service
        .register_staging_multipart(
            &loser.attempt_id,
            &StagingMultipart {
                target_store: multipart_identity.target_store_id.clone(),
                bucket: multipart_identity.bucket.clone(),
                key: multipart_identity.key.clone(),
                upload_id: multipart_identity.upload_id.clone(),
            },
        )
        .await
        .unwrap();
    service
        .publish_failure(
            &loser,
            &unique_id("failure"),
            RuntimeFailureClass::Authorization,
        )
        .await
        .unwrap();

    let winner_job = harness.job().await;
    let winner = harness.executing(&service, &winner_job).await;
    let winner_key =
        attempt_manifest_key_in("users/recovery", &winner.attempt_id, &unique_id("manifest"));
    let winner_object = harness
        .target
        .put_create_only(
            &CreateObjectRequest::parse(winner_key).unwrap(),
            b"winner-manifest".as_slice(),
        )
        .await
        .unwrap();
    service
        .register_staging_object(
            &winner.attempt_id,
            &StagingObject {
                target_store: winner_object.identity.target_store_id.clone(),
                bucket: winner_object.identity.bucket.clone(),
                key: winner_object.identity.key.clone(),
                version_id: winner_object.identity.version_id.clone(),
            },
        )
        .await
        .unwrap();
    let publication = harness.seal_result(&winner, winner_object.clone()).await;
    service
        .publish_success(&winner, &unique_id("success"), publication)
        .await
        .unwrap();

    let config = recovery_config();
    let first = service.claim_cleanup(&config).await.unwrap().unwrap();
    let failing_target = TargetStore::connect(
        TargetStoreConfig::new_exclusive(
            harness.target_config.target_store_id.clone(),
            "http://127.0.0.1:9",
            harness.target_config.bucket.clone(),
            harness.target_config.region.clone(),
            harness.target_config.access_key_id.clone(),
            harness.target_config.secret_access_key.clone(),
            "",
        )
        .unwrap(),
    )
    .unwrap();
    assert!(execute_cleanup(&failing_target, &first).await.is_err());
    assert_eq!(
        service
            .retry_cleanup(&first, "target_unavailable")
            .await
            .unwrap(),
        CleanupCompletion::RetryScheduled
    );
    let short_claims = RecoveryConfig {
        cleanup_claim_ttl: Duration::from_millis(1),
        ..config.clone()
    };
    let abandoned = service.claim_cleanup(&short_claims).await.unwrap().unwrap();
    tokio::time::sleep(Duration::from_millis(10)).await;
    let restarted = harness.service(3);
    let reclaimed = restarted.claim_cleanup(&config).await.unwrap().unwrap();
    assert_eq!(reclaimed.cleanup_id, abandoned.cleanup_id);
    assert_ne!(reclaimed.claim_token, abandoned.claim_token);
    assert_eq!(
        service.complete_cleanup(&abandoned).await.unwrap(),
        CleanupCompletion::ClaimLost
    );
    execute_cleanup(&harness.target, &reclaimed).await.unwrap();
    assert_eq!(
        restarted.complete_cleanup(&reclaimed).await.unwrap(),
        CleanupCompletion::Completed
    );
    let mut completed = 1;
    while let Some(claim) = restarted.claim_cleanup(&config).await.unwrap() {
        assert_eq!(claim.attempt_id, loser.attempt_id);
        execute_cleanup(&harness.target, &claim).await.unwrap();
        assert_eq!(
            restarted.complete_cleanup(&claim).await.unwrap(),
            CleanupCompletion::Completed
        );
        assert_eq!(
            restarted.complete_cleanup(&claim).await.unwrap(),
            CleanupCompletion::ClaimLost
        );
        completed += 1;
    }
    assert_eq!(completed, 2);
    assert!(
        harness
            .target
            .head_exact(&loser_object.identity)
            .await
            .is_err()
    );
    assert!(
        harness
            .target
            .head_exact(&winner_object.identity)
            .await
            .is_ok()
    );
    assert!(
        service
            .register_staging_object(
                &loser.attempt_id,
                &StagingObject {
                    target_store: harness.target_config.target_store_id.clone(),
                    bucket: harness.target_config.bucket.clone(),
                    key: attempt_object_key_in(
                        "users/recovery",
                        &winner.attempt_id,
                        &unique_id("wrong"),
                        None,
                    ),
                    version_id: None,
                },
            )
            .await
            .is_err()
    );
}

#[tokio::test]
async fn cleanup_reconciles_accepted_writes_from_durable_key_intents() {
    let Some(harness) = Harness::new("write-intent").await else {
        return;
    };
    let service = harness.service(1);
    let config = recovery_config();

    let multipart_job = harness.job().await;
    let multipart_attempt = harness.executing(&service, &multipart_job).await;
    let multipart_key = attempt_object_key_in(
        "users/recovery",
        &multipart_attempt.attempt_id,
        &unique_id("multipart"),
        None,
    );
    service
        .register_staging_write_intent_operation(
            &multipart_attempt,
            &unique_id("intent"),
            StagingWriteIntent {
                target_store: harness.target_config.target_store_id.clone(),
                bucket: harness.target_config.bucket.clone(),
                key: multipart_key.clone(),
            },
        )
        .await
        .unwrap();
    let accepted_multipart = harness
        .target
        .begin_multipart(&CreateObjectRequest::parse(multipart_key.clone()).unwrap())
        .await
        .unwrap();
    let accepted_upload_id = accepted_multipart.identity().upload_id.clone();
    service
        .publish_failure(
            &multipart_attempt,
            &unique_id("failure"),
            RuntimeFailureClass::Authorization,
        )
        .await
        .unwrap();
    let intent_claim = service.claim_cleanup(&config).await.unwrap().unwrap();
    assert!(matches!(
        intent_claim.identity,
        CleanupIdentity::WriteIntent(_)
    ));
    let discovered = harness
        .target
        .discover_staging_write(&multipart_key)
        .await
        .unwrap();
    assert!(discovered.objects.is_empty());
    assert_eq!(discovered.multiparts.len(), 1);
    assert_eq!(discovered.multiparts[0].upload_id, accepted_upload_id);
    assert_eq!(
        service
            .reconcile_write_intent(
                &intent_claim,
                Vec::new(),
                discovered
                    .multiparts
                    .into_iter()
                    .map(|identity| StagingMultipart {
                        target_store: identity.target_store_id,
                        bucket: identity.bucket,
                        key: identity.key,
                        upload_id: identity.upload_id,
                    })
                    .collect(),
            )
            .await
            .unwrap(),
        CleanupCompletion::Completed
    );
    let multipart_claim = service.claim_cleanup(&config).await.unwrap().unwrap();
    assert!(matches!(
        multipart_claim.identity,
        CleanupIdentity::Multipart(_)
    ));
    execute_cleanup(&harness.target, &multipart_claim)
        .await
        .unwrap();
    service.complete_cleanup(&multipart_claim).await.unwrap();
    assert!(
        harness
            .target
            .discover_staging_write(&multipart_key)
            .await
            .unwrap()
            .multiparts
            .is_empty()
    );

    let object_job = harness.job().await;
    let object_attempt = harness.executing(&service, &object_job).await;
    let object_key = attempt_object_key_in(
        "users/recovery",
        &object_attempt.attempt_id,
        &unique_id("object"),
        None,
    );
    service
        .register_staging_write_intent_operation(
            &object_attempt,
            &unique_id("intent"),
            StagingWriteIntent {
                target_store: harness.target_config.target_store_id.clone(),
                bucket: harness.target_config.bucket.clone(),
                key: object_key.clone(),
            },
        )
        .await
        .unwrap();
    let accepted_object = harness
        .target
        .put_create_only(
            &CreateObjectRequest::parse(object_key.clone()).unwrap(),
            b"accepted-before-ledger".as_slice(),
        )
        .await
        .unwrap();
    service
        .publish_failure(
            &object_attempt,
            &unique_id("failure"),
            RuntimeFailureClass::Authorization,
        )
        .await
        .unwrap();
    let intent_claim = service.claim_cleanup(&config).await.unwrap().unwrap();
    let discovered = harness
        .target
        .discover_staging_write(&object_key)
        .await
        .unwrap();
    let [discovered_object] = discovered.objects.as_slice() else {
        panic!("expected one accepted object");
    };
    assert_eq!(
        discovered_object.target_store_id,
        accepted_object.identity.target_store_id
    );
    assert_eq!(discovered_object.bucket, accepted_object.identity.bucket);
    assert_eq!(discovered_object.key, accepted_object.identity.key);
    if let (Some(discovered_version), Some(accepted_version)) = (
        discovered_object.version_id.as_deref(),
        accepted_object.identity.version_id.as_deref(),
    ) {
        assert_eq!(discovered_version, accepted_version);
    }
    assert!(discovered.multiparts.is_empty());
    assert_eq!(
        service
            .reconcile_write_intent(
                &intent_claim,
                discovered
                    .objects
                    .into_iter()
                    .map(|identity| StagingObject {
                        target_store: identity.target_store_id,
                        bucket: identity.bucket,
                        key: identity.key,
                        version_id: identity.version_id,
                    })
                    .collect(),
                Vec::new(),
            )
            .await
            .unwrap(),
        CleanupCompletion::Completed
    );
    let object_claim = service.claim_cleanup(&config).await.unwrap().unwrap();
    assert!(matches!(object_claim.identity, CleanupIdentity::Object(_)));
    execute_cleanup(&harness.target, &object_claim)
        .await
        .unwrap();
    service.complete_cleanup(&object_claim).await.unwrap();
    assert!(
        harness
            .target
            .head_exact(&accepted_object.identity)
            .await
            .is_err()
    );
}

async fn execute_cleanup(
    target: &TargetStore,
    claim: &remora_attempt::CleanupClaim,
) -> Result<(), remora_target::TargetError> {
    match &claim.identity {
        CleanupIdentity::WriteIntent(_) => Err(remora_target::TargetError::internal(
            "write intents must be reconciled before exact cleanup",
        )),
        CleanupIdentity::Object(object) => {
            target
                .delete_exact(&ObjectIdentity {
                    target_store_id: object.target_store.clone(),
                    bucket: object.bucket.clone(),
                    key: object.key.clone(),
                    version_id: object.version_id.clone(),
                })
                .await
        }
        CleanupIdentity::Multipart(upload) => {
            let (session, _) = target
                .resume_multipart(
                    MultipartIdentity {
                        target_store_id: upload.target_store.clone(),
                        bucket: upload.bucket.clone(),
                        key: upload.key.clone(),
                        upload_id: upload.upload_id.clone(),
                    },
                    None,
                )
                .await?;
            target.abort_multipart(session).await
        }
    }
}

fn recovery_config() -> RecoveryConfig {
    RecoveryConfig {
        retry_delay: Duration::ZERO,
        cleanup_claim_ttl: Duration::from_secs(10),
        batch_size: 128,
    }
}

fn required() -> bool {
    std::env::var("REMORA_RECOVERY_REQUIRE").is_ok()
}

fn required_env(name: &str) -> Option<String> {
    match std::env::var(name) {
        Ok(value) if !value.is_empty() => Some(value),
        Ok(_) | Err(_) if required() => panic!("{name} is required"),
        Ok(_) | Err(_) => None,
    }
}

fn unique_id(prefix: &str) -> OpaqueId {
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("time")
        .as_nanos();
    OpaqueId::parse(format!("{prefix}{nanos:x}")).unwrap()
}

fn nonce() -> u128 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("time")
        .as_nanos()
}

fn coordination_namespace() -> String {
    std::env::var("REMORA_COORDINATION_NAMESPACE").unwrap_or_else(|_| "remora:test:".into())
}

fn coordination_upstream() -> String {
    std::env::var("REMORA_COORDINATION_URL")
        .unwrap()
        .trim_start_matches("redis://")
        .split('/')
        .next()
        .unwrap()
        .to_owned()
}