use std::time::Duration;

use futures_util::TryStreamExt;
use remora_domain::{BucketName, OpaqueId, Target, TargetPrefix, TargetStoreId};
use remora_durable::{PgTx, enqueue};
use remora_identity::{
    AuthorizedResource, ManagementOperation, Principal, ResourceVisibility, resource_visibility,
};
use remora_target::{
    MultipartIdentity, ObjectIdentity, TargetError, attempt_key_attempt_id,
    attempt_manifest_key_in, attempt_object_prefix, attempt_staging_prefix, validate_attempt_key,
};
use serde::Deserialize;
use serde_json::{Value, json};
use sqlx::{FromRow, PgPool, Postgres, Transaction};
use time::OffsetDateTime;

use crate::{
    DeleteClaim, DeleteIdentity, DeletionConfig, DeletionError, DeletionOperation, DeletionProblem,
    DeletionProblemClass, DeletionState, DrainOutcome, HardDeleteOutcome, LifecycleMutation,
    TargetRegistry,
};

#[derive(Clone)]
pub struct DeletionService {
    pool: PgPool,
    targets: TargetRegistry,
    config: DeletionConfig,
}

impl DeletionService {
    pub fn new(
        pool: PgPool,
        targets: TargetRegistry,
        config: DeletionConfig,
    ) -> Result<Self, DeletionError> {
        if config.claim_ttl.is_zero() {
            return Err(DeletionError::InvalidConfig(
                "claim ttl must be greater than zero".into(),
            ));
        }
        Ok(Self {
            pool,
            targets,
            config,
        })
    }

    pub async fn soft_delete(
        &self,
        principal: &Principal,
        job_id: &OpaqueId,
    ) -> Result<LifecycleMutation, DeletionError> {
        let mut tx = self.pool.begin().await?;
        let outcome = Self::soft_delete_in_tx(&mut tx, principal, job_id).await?;
        tx.commit().await?;
        Ok(outcome)
    }

    pub async fn soft_delete_in_tx(
        tx: &mut PgTx<'_>,
        principal: &Principal,
        job_id: &OpaqueId,
    ) -> Result<LifecycleMutation, DeletionError> {
        let row = lock_job(tx, job_id).await?;
        authorize(principal, &row, ManagementOperation::SoftDelete)?;
        require_terminal(&row.phase)?;
        if row.deletion_operation_id.is_some() {
            return Err(DeletionError::HardDeleteStarted);
        }
        if row.lifecycle == "soft_deleted" {
            return Ok(LifecycleMutation::Unchanged);
        }
        let now = OffsetDateTime::now_utc();
        sqlx::query(
            "UPDATE jobs
             SET lifecycle = 'soft_deleted', revision = revision + 1, updated_at = $2
             WHERE job_id = $1",
        )
        .bind(job_id.as_str())
        .bind(now)
        .execute(tx.as_mut())
        .await?;
        enqueue(tx, "job.updated", json!({ "job_id": job_id })).await?;
        Ok(LifecycleMutation::Changed)
    }

    pub async fn restore(
        &self,
        principal: &Principal,
        job_id: &OpaqueId,
    ) -> Result<LifecycleMutation, DeletionError> {
        let mut tx = self.pool.begin().await?;
        let outcome = Self::restore_in_tx(&mut tx, principal, job_id).await?;
        tx.commit().await?;
        Ok(outcome)
    }

    pub async fn restore_in_tx(
        tx: &mut PgTx<'_>,
        principal: &Principal,
        job_id: &OpaqueId,
    ) -> Result<LifecycleMutation, DeletionError> {
        let row = lock_job(tx, job_id).await?;
        authorize(principal, &row, ManagementOperation::Restore)?;
        require_terminal(&row.phase)?;
        if row.deletion_operation_id.is_some() {
            return Err(DeletionError::HardDeleteStarted);
        }
        if row.lifecycle == "visible" {
            return Ok(LifecycleMutation::Unchanged);
        }
        let now = OffsetDateTime::now_utc();
        sqlx::query(
            "UPDATE jobs
             SET lifecycle = 'visible', revision = revision + 1, updated_at = $2
             WHERE job_id = $1",
        )
        .bind(job_id.as_str())
        .bind(now)
        .execute(tx.as_mut())
        .await?;
        enqueue(tx, "job.updated", json!({ "job_id": job_id })).await?;
        Ok(LifecycleMutation::Changed)
    }

    pub async fn hard_delete(
        &self,
        principal: &Principal,
        job_id: &OpaqueId,
    ) -> Result<HardDeleteOutcome, DeletionError> {
        let mut tx = self.pool.begin().await?;
        let outcome = Self::hard_delete_in_tx(&mut tx, principal, job_id).await?;
        tx.commit().await?;
        Ok(outcome)
    }

    pub async fn hard_delete_in_tx(
        tx: &mut PgTx<'_>,
        principal: &Principal,
        job_id: &OpaqueId,
    ) -> Result<HardDeleteOutcome, DeletionError> {
        let operation_id = new_id('d')?;
        let row = lock_job(tx, job_id).await?;
        authorize(principal, &row, ManagementOperation::HardDelete)?;
        require_terminal(&row.phase)?;
        if let Some(existing) = row.deletion_operation_id {
            let operation = load_operation_in_tx(tx, &existing).await?;
            return Ok(HardDeleteOutcome {
                mutation: LifecycleMutation::Unchanged,
                operation,
            });
        }
        validate_committed_inventory(tx, &row).await?;
        validate_deletion_identities(tx, &row).await?;
        let now = OffsetDateTime::now_utc();
        sqlx::query(
            "INSERT INTO deletion_operations (
                operation_id, job_id, state, created_at, updated_at
             ) VALUES ($1, $2, 'queued', $3, $3)",
        )
        .bind(operation_id.as_str())
        .bind(job_id.as_str())
        .bind(now)
        .execute(tx.as_mut())
        .await?;
        sqlx::query(
            "UPDATE jobs
             SET lifecycle = 'hard_deleting', deletion_operation_id = $2,
                 hard_delete_started_at = $3, revision = revision + 1, updated_at = $3
             WHERE job_id = $1",
        )
        .bind(job_id.as_str())
        .bind(operation_id.as_str())
        .bind(now)
        .execute(tx.as_mut())
        .await?;
        materialize_items(tx, job_id, &operation_id, now).await?;
        finalize_if_empty(tx, job_id, &operation_id, now).await?;
        enqueue(
            tx,
            "job.updated",
            json!({ "job_id": job_id, "deletion_operation_id": operation_id }),
        )
        .await?;
        enqueue(
            tx,
            "deletion.updated",
            json!({ "job_id": job_id, "operation_id": operation_id }),
        )
        .await?;
        let operation = load_operation_in_tx(tx, operation_id.as_str()).await?;
        Ok(HardDeleteOutcome {
            mutation: LifecycleMutation::Changed,
            operation,
        })
    }

    pub async fn operation(
        &self,
        principal: &Principal,
        operation_id: &OpaqueId,
    ) -> Result<DeletionOperation, DeletionError> {
        let row = sqlx::query_as::<_, OperationAccessRow>(
            "SELECT o.operation_id, o.job_id, o.state, o.problem_class, o.problem_message,
                    o.created_at, o.updated_at, o.completed_at,
                    j.owner_id, j.target_store, j.target_bucket, j.target_prefix
             FROM deletion_operations o
             JOIN jobs j ON j.job_id = o.job_id
             WHERE o.operation_id = $1",
        )
        .bind(operation_id.as_str())
        .fetch_optional(&self.pool)
        .await?
        .ok_or(DeletionError::NotFound)?;
        authorize_parts(
            principal,
            &row.owner_id,
            &row.target_store,
            &row.target_bucket,
            &row.target_prefix,
            ManagementOperation::Read,
        )?;
        row.into_operation()
    }

    pub async fn claim_next(
        &self,
        operation_id: &OpaqueId,
    ) -> Result<Option<DeleteClaim>, DeletionError> {
        let claim_token = new_id('c')?;
        let claim_seconds = duration_seconds(self.config.claim_ttl)?;
        let mut tx = self.pool.begin().await?;
        let row = sqlx::query_as::<_, ClaimRow>(
            "WITH candidate AS (
                SELECT i.item_id
                FROM deletion_items i
                JOIN deletion_operations o ON o.operation_id = i.operation_id
                WHERE i.operation_id = $1
                  AND o.state <> 'completed'
                  AND (i.state = 'pending'
                       OR (i.state = 'claimed' AND i.claimed_until <= now()))
                  AND (
                      i.kind <> 'manifest'
                      OR NOT EXISTS (
                          SELECT 1 FROM deletion_items prior
                          WHERE prior.operation_id = i.operation_id
                            AND prior.kind <> 'manifest'
                            AND prior.state <> 'complete'
                      )
                  )
                ORDER BY i.priority, i.item_id
                LIMIT 1
                FOR UPDATE OF i SKIP LOCKED
             )
             UPDATE deletion_items i
             SET state = 'claimed', claim_token = $2,
                 claimed_until = now() + ($3 * interval '1 second')
             FROM candidate
             WHERE i.item_id = candidate.item_id
             RETURNING i.item_id, i.operation_id, i.kind, i.target_store, i.bucket,
                       i.object_key, i.version_id, i.upload_id, i.claim_token",
        )
        .bind(operation_id.as_str())
        .bind(claim_token.as_str())
        .bind(claim_seconds)
        .fetch_optional(tx.as_mut())
        .await?;
        if row.is_some() {
            let job_id: Option<String> = sqlx::query_scalar(
                "UPDATE deletion_operations
                 SET state = 'running', problem_class = NULL, problem_message = NULL,
                     updated_at = now()
                 WHERE operation_id = $1 AND state NOT IN ('running', 'completed')
                 RETURNING job_id",
            )
            .bind(operation_id.as_str())
            .fetch_optional(tx.as_mut())
            .await?;
            if let Some(job_id) = job_id {
                enqueue(
                    &mut tx,
                    "deletion.updated",
                    json!({ "job_id": job_id, "operation_id": operation_id }),
                )
                .await?;
            }
        }
        tx.commit().await?;
        row.map(ClaimRow::into_claim).transpose()
    }

    pub async fn process_next(
        &self,
        operation_id: &OpaqueId,
    ) -> Result<DrainOutcome, DeletionError> {
        let Some(claim) = self.claim_next(operation_id).await? else {
            return self.finish_or_idle(operation_id).await;
        };
        let (target_store, bucket) = identity_target(&claim.identity);
        let Some(target) = self.targets.resolve(target_store, bucket) else {
            self.block_claim(
                &claim,
                DeletionProblemClass::Configuration,
                "target configuration is unavailable",
            )
            .await?;
            return Ok(DrainOutcome::Blocked);
        };
        let effect = match &claim.identity {
            DeleteIdentity::Object(identity) | DeleteIdentity::Manifest(identity) => {
                target.delete_exact(identity).await
            }
            DeleteIdentity::Multipart(identity) => target.abort_multipart_exact(identity).await,
            DeleteIdentity::WriteIntent(identity) => {
                self.reconcile_write_intent(&claim, &target, identity).await
            }
        };
        match effect {
            Ok(()) => match self.complete_claim(&claim).await {
                Err(DeletionError::ClaimLost) => Ok(DrainOutcome::Idle),
                outcome => outcome,
            },
            Err(error) => {
                let (class, message) = target_problem(&error);
                match self.block_claim(&claim, class, message).await {
                    Ok(()) => Ok(DrainOutcome::Blocked),
                    Err(DeletionError::ClaimLost) => Ok(DrainOutcome::Idle),
                    Err(error) => Err(error),
                }
            }
        }
    }

    async fn reconcile_write_intent(
        &self,
        claim: &DeleteClaim,
        target: &remora_target::TargetStore,
        identity: &ObjectIdentity,
    ) -> Result<(), TargetError> {
        let discovered = target.discover_staging_write(&identity.key).await?;
        let mut tx = self.pool.begin().await.map_err(database_target_error)?;
        let authoritative: bool = sqlx::query_scalar(
            "SELECT EXISTS (
                SELECT 1 FROM deletion_items
                WHERE item_id = $1 AND state = 'claimed' AND claim_token = $2
             )",
        )
        .bind(i64::try_from(claim.item_id).map_err(|_| TargetError::internal("invalid item id"))?)
        .bind(claim.claim_token.as_str())
        .fetch_one(tx.as_mut())
        .await
        .map_err(database_target_error)?;
        if !authoritative {
            return Err(TargetError::internal("deletion claim expired"));
        }
        let now = OffsetDateTime::now_utc();
        for object in discovered.objects {
            insert_discovered_object(&mut tx, &claim.operation_id, &object, now)
                .await
                .map_err(database_target_error)?;
        }
        for multipart in discovered.multiparts {
            insert_discovered_multipart(&mut tx, &claim.operation_id, &multipart, now)
                .await
                .map_err(database_target_error)?;
        }
        tx.commit().await.map_err(database_target_error)?;
        Ok(())
    }

    async fn complete_claim(&self, claim: &DeleteClaim) -> Result<DrainOutcome, DeletionError> {
        let item_id = i64::try_from(claim.item_id).map_err(|_| DeletionError::ClaimLost)?;
        let mut tx = self.pool.begin().await?;
        let now = OffsetDateTime::now_utc();
        let changed = sqlx::query(
            "UPDATE deletion_items
             SET state = 'complete', claim_token = NULL, claimed_until = NULL, completed_at = $3
             WHERE item_id = $1 AND state = 'claimed' AND claim_token = $2",
        )
        .bind(item_id)
        .bind(claim.claim_token.as_str())
        .bind(now)
        .execute(tx.as_mut())
        .await?;
        if changed.rows_affected() != 1 {
            return Err(DeletionError::ClaimLost);
        }
        let job_id: String = sqlx::query_scalar(
            "SELECT job_id FROM deletion_operations WHERE operation_id = $1 FOR UPDATE",
        )
        .bind(claim.operation_id.as_str())
        .fetch_one(tx.as_mut())
        .await?;
        let completed = finalize_if_empty(
            &mut tx,
            &OpaqueId::parse(&job_id).map_err(durable_effect)?,
            &claim.operation_id,
            now,
        )
        .await?;
        if !completed {
            sqlx::query(
                "UPDATE deletion_operations
                 SET state = 'queued', problem_class = NULL, problem_message = NULL,
                     updated_at = $2
                 WHERE operation_id = $1",
            )
            .bind(claim.operation_id.as_str())
            .bind(now)
            .execute(tx.as_mut())
            .await?;
            enqueue(
                &mut tx,
                "deletion.updated",
                json!({ "job_id": job_id, "operation_id": claim.operation_id }),
            )
            .await?;
        }
        tx.commit().await?;
        Ok(if completed {
            DrainOutcome::Completed
        } else {
            DrainOutcome::ItemCompleted
        })
    }

    async fn block_claim(
        &self,
        claim: &DeleteClaim,
        class: DeletionProblemClass,
        message: &str,
    ) -> Result<(), DeletionError> {
        let item_id = i64::try_from(claim.item_id).map_err(|_| DeletionError::ClaimLost)?;
        let mut tx = self.pool.begin().await?;
        let changed = sqlx::query(
            "UPDATE deletion_items
             SET state = 'pending', claim_token = NULL, claimed_until = NULL,
                 failure_count = failure_count + 1
             WHERE item_id = $1 AND state = 'claimed' AND claim_token = $2",
        )
        .bind(item_id)
        .bind(claim.claim_token.as_str())
        .execute(tx.as_mut())
        .await?;
        if changed.rows_affected() != 1 {
            return Err(DeletionError::ClaimLost);
        }
        sqlx::query(
            "UPDATE deletion_operations
             SET state = 'blocked', problem_class = $2, problem_message = $3, updated_at = now()
             WHERE operation_id = $1 AND state <> 'completed'",
        )
        .bind(claim.operation_id.as_str())
        .bind(problem_class_name(class))
        .bind(message)
        .execute(tx.as_mut())
        .await?;
        let job_id: String =
            sqlx::query_scalar("SELECT job_id FROM deletion_operations WHERE operation_id = $1")
                .bind(claim.operation_id.as_str())
                .fetch_one(tx.as_mut())
                .await?;
        enqueue(
            &mut tx,
            "deletion.updated",
            json!({ "job_id": job_id, "operation_id": claim.operation_id }),
        )
        .await?;
        tx.commit().await?;
        Ok(())
    }

    async fn finish_or_idle(&self, operation_id: &OpaqueId) -> Result<DrainOutcome, DeletionError> {
        let mut tx = self.pool.begin().await?;
        let row = sqlx::query_as::<_, (String, String)>(
            "SELECT job_id, state FROM deletion_operations
             WHERE operation_id = $1 FOR UPDATE",
        )
        .bind(operation_id.as_str())
        .fetch_optional(tx.as_mut())
        .await?
        .ok_or(DeletionError::NotFound)?;
        if row.1 == "completed" {
            tx.commit().await?;
            return Ok(DrainOutcome::Completed);
        }
        let job_id = OpaqueId::parse(row.0).map_err(durable_effect)?;
        let completed =
            finalize_if_empty(&mut tx, &job_id, operation_id, OffsetDateTime::now_utc()).await?;
        tx.commit().await?;
        Ok(if completed {
            DrainOutcome::Completed
        } else {
            DrainOutcome::Idle
        })
    }
}

async fn lock_job(
    tx: &mut Transaction<'_, Postgres>,
    job_id: &OpaqueId,
) -> Result<JobRow, DeletionError> {
    sqlx::query_as::<_, JobRow>(
        "SELECT job_id, owner_id, phase, lifecycle, target_store, target_bucket,
                target_prefix, committed_result, deletion_operation_id
         FROM jobs WHERE job_id = $1 FOR UPDATE",
    )
    .bind(job_id.as_str())
    .fetch_optional(tx.as_mut())
    .await?
    .ok_or(DeletionError::NotFound)
}

fn authorize(
    principal: &Principal,
    row: &JobRow,
    operation: ManagementOperation,
) -> Result<(), DeletionError> {
    authorize_parts(
        principal,
        &row.owner_id,
        &row.target_store,
        &row.target_bucket,
        &row.target_prefix,
        operation,
    )
}

fn authorize_parts(
    principal: &Principal,
    owner: &str,
    target_store: &str,
    bucket: &str,
    prefix: &Value,
    operation: ManagementOperation,
) -> Result<(), DeletionError> {
    let resource = AuthorizedResource {
        owner: OpaqueId::parse(owner).map_err(durable_effect)?,
        target: parse_target(target_store, bucket, prefix)?,
    };
    match resource_visibility(principal, Some(&resource), operation) {
        ResourceVisibility::Visible => Ok(()),
        ResourceVisibility::NotFound => Err(DeletionError::NotFound),
    }
}

fn parse_target(store: &str, bucket: &str, prefix: &Value) -> Result<Target, DeletionError> {
    let segments: Vec<String> =
        serde_json::from_value(prefix.clone()).map_err(|_| DeletionError::CorruptInventory)?;
    Ok(Target {
        store: TargetStoreId::parse(store).map_err(durable_effect)?,
        bucket: BucketName::parse(bucket).map_err(durable_effect)?,
        prefix: TargetPrefix::parse(&segments.join("/"), std::iter::empty::<&str>())
            .map_err(durable_effect)?,
    })
}

fn require_terminal(phase: &str) -> Result<(), DeletionError> {
    if matches!(phase, "succeeded" | "failed" | "cancelled") {
        Ok(())
    } else {
        Err(DeletionError::NotTerminal)
    }
}

async fn validate_committed_inventory(
    tx: &mut Transaction<'_, Postgres>,
    job: &JobRow,
) -> Result<(), DeletionError> {
    if job.phase != "succeeded" {
        return if job.committed_result.is_none() {
            Ok(())
        } else {
            Err(DeletionError::CorruptInventory)
        };
    }
    let committed: CommittedResult = serde_json::from_value(
        job.committed_result
            .clone()
            .ok_or(DeletionError::CorruptInventory)?,
    )
    .map_err(|_| DeletionError::CorruptInventory)?;
    if committed.schema_version != 1 {
        return Err(DeletionError::CorruptInventory);
    }
    let row = sqlx::query_as::<_, InventoryValidationRow>(
        "SELECT r.state, r.manifest_schema_version, r.manifest_id, r.manifest_store,
                r.manifest_bucket, r.manifest_key, r.manifest_version_id,
                r.manifest_sha256, r.manifest_size, r.member_count, r.total_bytes,
                r.staging_origin_attempt_id, r.target_prefix,
                committed_attempt.job_id AS committed_attempt_job_id,
                staging_origin.job_id AS staging_origin_job_id,
                count(m.member_id) AS actual_member_count,
                COALESCE(sum(m.size), 0)::BIGINT AS actual_total_bytes
         FROM result_inventories r
         LEFT JOIN attempts committed_attempt ON committed_attempt.attempt_id = r.attempt_id
         LEFT JOIN attempts staging_origin
                ON staging_origin.attempt_id = r.staging_origin_attempt_id
         LEFT JOIN result_inventory_members m ON m.attempt_id = r.attempt_id
         WHERE r.attempt_id = $1
         GROUP BY r.attempt_id, committed_attempt.job_id, staging_origin.job_id",
    )
    .bind(committed.attempt_id.as_str())
    .fetch_optional(tx.as_mut())
    .await?
    .ok_or(DeletionError::CorruptInventory)?;
    let manifest_size =
        i64::try_from(committed.manifest.size).map_err(|_| DeletionError::CorruptInventory)?;
    let member_count =
        i64::try_from(committed.member_count).map_err(|_| DeletionError::CorruptInventory)?;
    let total_bytes =
        i64::try_from(committed.total_bytes).map_err(|_| DeletionError::CorruptInventory)?;
    let schema_version =
        i32::try_from(committed.schema_version).map_err(|_| DeletionError::CorruptInventory)?;
    let target_prefix = canonical_target_prefix(&job.target_prefix)?;
    let staging_origin = row
        .staging_origin_attempt_id
        .as_deref()
        .ok_or(DeletionError::CorruptInventory)
        .and_then(|value| OpaqueId::parse(value).map_err(|_| DeletionError::CorruptInventory))?;
    let expected_manifest_key =
        attempt_manifest_key_in(&target_prefix, &staging_origin, &committed.manifest_id);
    let matches = row.state == "sealed"
        && row.manifest_schema_version == Some(schema_version)
        && row.manifest_id.as_deref() == Some(committed.manifest_id.as_str())
        && row.manifest_store.as_deref() == Some(committed.manifest.store.as_str())
        && row.manifest_bucket.as_deref() == Some(committed.manifest.bucket.as_str())
        && row.manifest_key.as_deref() == Some(committed.manifest.key.as_str())
        && row.manifest_version_id == committed.manifest.version_id
        && row.manifest_sha256.as_deref() == Some(&committed.manifest.sha256)
        && row.manifest_size == Some(manifest_size)
        && row.member_count == member_count
        && row.total_bytes == total_bytes
        && row.actual_member_count == member_count
        && row.actual_total_bytes == total_bytes
        && row.committed_attempt_job_id.as_deref() == Some(job.job_id.as_str())
        && row.staging_origin_job_id.as_deref() == Some(job.job_id.as_str())
        && row.target_prefix.as_deref() == Some(target_prefix.as_str())
        && committed.manifest.store == job.target_store
        && committed.manifest.bucket == job.target_bucket
        && committed.manifest.key == expected_manifest_key;
    if matches {
        Ok(())
    } else {
        Err(DeletionError::CorruptInventory)
    }
}

async fn validate_deletion_identities(
    tx: &mut Transaction<'_, Postgres>,
    job: &JobRow,
) -> Result<(), DeletionError> {
    let target_prefix = canonical_target_prefix(&job.target_prefix)?;
    let mut rows = sqlx::query_as::<_, OwnedIdentityRow>(
        "SELECT 'member' AS kind, m.object_store AS target_store,
                m.object_bucket AS bucket, m.object_key,
                r.staging_origin_attempt_id, r.target_prefix,
                NULL::TEXT AS manifest_id, origin.job_id AS origin_job_id
         FROM result_inventory_members m
         JOIN result_inventories r ON r.attempt_id = m.attempt_id
         JOIN attempts a ON a.attempt_id = m.attempt_id
         LEFT JOIN attempts origin ON origin.attempt_id = r.staging_origin_attempt_id
         WHERE a.job_id = $1
         UNION ALL
         SELECT 'manifest', r.manifest_store, r.manifest_bucket, r.manifest_key,
                r.staging_origin_attempt_id, r.target_prefix, r.manifest_id,
                origin.job_id
         FROM result_inventories r
         JOIN attempts a ON a.attempt_id = r.attempt_id
         LEFT JOIN attempts origin ON origin.attempt_id = r.staging_origin_attempt_id
         WHERE a.job_id = $1 AND r.state = 'sealed'
         UNION ALL
         SELECT 'cleanup', c.target_store, c.bucket, c.object_key,
                a.staging_origin_attempt_id, $2::TEXT, NULL::TEXT, origin.job_id
         FROM attempt_staging_cleanup c
         JOIN attempts a ON a.attempt_id = c.attempt_id
         LEFT JOIN attempts origin ON origin.attempt_id = a.staging_origin_attempt_id
         WHERE a.job_id = $1
           AND (c.kind <> 'write_intent' OR c.state <> 'complete')",
    )
    .bind(&job.job_id)
    .bind(&target_prefix)
    .fetch(tx.as_mut());
    while let Some(row) = rows.try_next().await? {
        validate_owned_identity(job, &target_prefix, row)?;
    }
    Ok(())
}

fn validate_owned_identity(
    job: &JobRow,
    job_target_prefix: &str,
    row: OwnedIdentityRow,
) -> Result<(), DeletionError> {
    if row.target_store.as_deref() != Some(job.target_store.as_str())
        || row.bucket.as_deref() != Some(job.target_bucket.as_str())
        || row.target_prefix.as_deref() != Some(job_target_prefix)
        || row.origin_job_id.as_deref() != Some(job.job_id.as_str())
    {
        return Err(DeletionError::CorruptInventory);
    }
    let key = row.object_key.ok_or(DeletionError::CorruptInventory)?;
    validate_attempt_key(&key).map_err(|_| DeletionError::CorruptInventory)?;
    let origin = row
        .staging_origin_attempt_id
        .ok_or(DeletionError::CorruptInventory)
        .and_then(|value| OpaqueId::parse(value).map_err(|_| DeletionError::CorruptInventory))?;
    if attempt_key_attempt_id(&key).map_err(|_| DeletionError::CorruptInventory)? != origin {
        return Err(DeletionError::CorruptInventory);
    }
    let expected = match row.kind.as_str() {
        "member" => attempt_object_prefix(job_target_prefix, &origin),
        "cleanup" => attempt_staging_prefix(job_target_prefix, &origin),
        "manifest" => attempt_manifest_key_in(
            job_target_prefix,
            &origin,
            &OpaqueId::parse(row.manifest_id.ok_or(DeletionError::CorruptInventory)?)
                .map_err(|_| DeletionError::CorruptInventory)?,
        ),
        _ => return Err(DeletionError::CorruptInventory),
    };
    let matches = if row.kind == "manifest" {
        key == expected
    } else {
        key.starts_with(&expected)
    };
    if matches {
        Ok(())
    } else {
        Err(DeletionError::CorruptInventory)
    }
}

fn canonical_target_prefix(value: &Value) -> Result<String, DeletionError> {
    let segments: Vec<String> =
        serde_json::from_value(value.clone()).map_err(|_| DeletionError::CorruptInventory)?;
    let canonical = TargetPrefix::parse(&segments.join("/"), std::iter::empty::<&str>())
        .map_err(|_| DeletionError::CorruptInventory)?
        .canonical();
    if canonical.split('/').eq(segments.iter().map(String::as_str))
        || (canonical.is_empty() && segments.is_empty())
    {
        Ok(canonical)
    } else {
        Err(DeletionError::CorruptInventory)
    }
}

async fn materialize_items(
    tx: &mut Transaction<'_, Postgres>,
    job_id: &OpaqueId,
    operation_id: &OpaqueId,
    now: OffsetDateTime,
) -> Result<(), DeletionError> {
    sqlx::query(
        "INSERT INTO deletion_items (
            operation_id, kind, priority, target_store, bucket, object_key,
            version_id, upload_id, created_at
         )
         SELECT $2, 'object', 10, m.object_store, m.object_bucket, m.object_key,
                m.object_version_id, NULL, $3
         FROM result_inventory_members m
         JOIN attempts a ON a.attempt_id = m.attempt_id
         WHERE a.job_id = $1
         ON CONFLICT DO NOTHING",
    )
    .bind(job_id.as_str())
    .bind(operation_id.as_str())
    .bind(now)
    .execute(tx.as_mut())
    .await?;
    sqlx::query(
        "INSERT INTO deletion_items (
            operation_id, kind, priority, target_store, bucket, object_key,
            version_id, upload_id, created_at
         )
         SELECT $2, 'manifest', 100, r.manifest_store, r.manifest_bucket, r.manifest_key,
                r.manifest_version_id, NULL, $3
         FROM result_inventories r
         JOIN attempts a ON a.attempt_id = r.attempt_id
         WHERE a.job_id = $1 AND r.state = 'sealed'
         ON CONFLICT DO NOTHING",
    )
    .bind(job_id.as_str())
    .bind(operation_id.as_str())
    .bind(now)
    .execute(tx.as_mut())
    .await?;
    sqlx::query(
        "INSERT INTO deletion_items (
            operation_id, kind, priority, target_store, bucket, object_key,
            version_id, upload_id, created_at
         )
         SELECT $2, c.kind,
                CASE c.kind WHEN 'multipart' THEN 0 WHEN 'write_intent' THEN 0 ELSE 10 END,
                c.target_store, c.bucket, c.object_key, c.version_id, c.upload_id, $3
         FROM attempt_staging_cleanup c
         JOIN attempts a ON a.attempt_id = c.attempt_id
         WHERE a.job_id = $1
           AND (c.kind <> 'write_intent' OR c.state <> 'complete')
         ON CONFLICT DO NOTHING",
    )
    .bind(job_id.as_str())
    .bind(operation_id.as_str())
    .bind(now)
    .execute(tx.as_mut())
    .await?;
    Ok(())
}

async fn finalize_if_empty(
    tx: &mut Transaction<'_, Postgres>,
    job_id: &OpaqueId,
    operation_id: &OpaqueId,
    now: OffsetDateTime,
) -> Result<bool, DeletionError> {
    let remaining: bool = sqlx::query_scalar(
        "SELECT EXISTS (
            SELECT 1 FROM deletion_items
            WHERE operation_id = $1 AND state <> 'complete'
         )",
    )
    .bind(operation_id.as_str())
    .fetch_one(tx.as_mut())
    .await?;
    if remaining {
        return Ok(false);
    }
    let operation_changed = sqlx::query(
        "UPDATE deletion_operations
         SET state = 'completed', problem_class = NULL, problem_message = NULL,
             updated_at = $2, completed_at = $2
         WHERE operation_id = $1 AND state <> 'completed'",
    )
    .bind(operation_id.as_str())
    .bind(now)
    .execute(tx.as_mut())
    .await?;
    if operation_changed.rows_affected() == 1 {
        enqueue(
            tx,
            "deletion.updated",
            json!({ "job_id": job_id, "operation_id": operation_id }),
        )
        .await?;
    }
    let changed = sqlx::query(
        "UPDATE jobs
         SET lifecycle = 'hard_deleted', revision = revision + 1, updated_at = $3
         WHERE job_id = $1 AND deletion_operation_id = $2 AND lifecycle = 'hard_deleting'",
    )
    .bind(job_id.as_str())
    .bind(operation_id.as_str())
    .bind(now)
    .execute(tx.as_mut())
    .await?;
    if changed.rows_affected() == 1 {
        enqueue(
            tx,
            "job.updated",
            json!({ "job_id": job_id, "deletion_operation_id": operation_id }),
        )
        .await?;
    }
    Ok(true)
}

async fn load_operation_in_tx(
    tx: &mut Transaction<'_, Postgres>,
    operation_id: &str,
) -> Result<DeletionOperation, DeletionError> {
    sqlx::query_as::<_, OperationRow>(
        "SELECT operation_id, job_id, state, problem_class, problem_message,
                created_at, updated_at, completed_at
         FROM deletion_operations WHERE operation_id = $1",
    )
    .bind(operation_id)
    .fetch_optional(tx.as_mut())
    .await?
    .ok_or(DeletionError::NotFound)?
    .into_operation()
}

async fn insert_discovered_object(
    tx: &mut Transaction<'_, Postgres>,
    operation_id: &OpaqueId,
    object: &ObjectIdentity,
    now: OffsetDateTime,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        "INSERT INTO deletion_items (
            operation_id, kind, priority, target_store, bucket, object_key,
            version_id, created_at
         ) VALUES ($1, 'object', 10, $2, $3, $4, $5, $6)
         ON CONFLICT DO NOTHING",
    )
    .bind(operation_id.as_str())
    .bind(&object.target_store_id)
    .bind(&object.bucket)
    .bind(&object.key)
    .bind(&object.version_id)
    .bind(now)
    .execute(tx.as_mut())
    .await?;
    Ok(())
}

async fn insert_discovered_multipart(
    tx: &mut Transaction<'_, Postgres>,
    operation_id: &OpaqueId,
    multipart: &MultipartIdentity,
    now: OffsetDateTime,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        "INSERT INTO deletion_items (
            operation_id, kind, priority, target_store, bucket, object_key,
            upload_id, created_at
         ) VALUES ($1, 'multipart', 0, $2, $3, $4, $5, $6)
         ON CONFLICT DO NOTHING",
    )
    .bind(operation_id.as_str())
    .bind(&multipart.target_store_id)
    .bind(&multipart.bucket)
    .bind(&multipart.key)
    .bind(&multipart.upload_id)
    .bind(now)
    .execute(tx.as_mut())
    .await?;
    Ok(())
}

fn target_problem(error: &TargetError) -> (DeletionProblemClass, &'static str) {
    match error {
        TargetError::Authorization => (
            DeletionProblemClass::Authorization,
            "target authorization is unavailable",
        ),
        TargetError::Validation(_) => (
            DeletionProblemClass::Configuration,
            "target configuration is unavailable",
        ),
        TargetError::Transient(_) | TargetError::Capacity => (
            DeletionProblemClass::TargetUnavailable,
            "target service is unavailable",
        ),
        TargetError::NotFound | TargetError::AlreadyExists | TargetError::Integrity(_) => (
            DeletionProblemClass::Internal,
            "target deletion could not be completed",
        ),
        TargetError::Internal(_) => (
            DeletionProblemClass::Internal,
            "target deletion could not be completed",
        ),
    }
}

fn identity_target(identity: &DeleteIdentity) -> (&str, &str) {
    match identity {
        DeleteIdentity::WriteIntent(identity)
        | DeleteIdentity::Object(identity)
        | DeleteIdentity::Manifest(identity) => (&identity.target_store_id, &identity.bucket),
        DeleteIdentity::Multipart(identity) => (&identity.target_store_id, &identity.bucket),
    }
}

fn problem_class_name(class: DeletionProblemClass) -> &'static str {
    match class {
        DeletionProblemClass::Authorization => "authorization",
        DeletionProblemClass::TargetUnavailable => "target_unavailable",
        DeletionProblemClass::Configuration => "configuration",
        DeletionProblemClass::Internal => "internal",
    }
}

fn parse_problem_class(value: &str) -> Result<DeletionProblemClass, DeletionError> {
    match value {
        "authorization" => Ok(DeletionProblemClass::Authorization),
        "target_unavailable" => Ok(DeletionProblemClass::TargetUnavailable),
        "configuration" => Ok(DeletionProblemClass::Configuration),
        "internal" => Ok(DeletionProblemClass::Internal),
        _ => Err(DeletionError::CorruptInventory),
    }
}

fn parse_state(value: &str) -> Result<DeletionState, DeletionError> {
    match value {
        "queued" => Ok(DeletionState::Queued),
        "running" => Ok(DeletionState::Running),
        "retry_wait" => Ok(DeletionState::RetryWait),
        "blocked" => Ok(DeletionState::Blocked),
        "completed" => Ok(DeletionState::Completed),
        _ => Err(DeletionError::CorruptInventory),
    }
}

fn new_id(prefix: char) -> Result<OpaqueId, DeletionError> {
    let mut bytes = [0_u8; 16];
    getrandom::fill(&mut bytes).map_err(|error| {
        DeletionError::InvalidConfig(format!("failed to draw identifier entropy: {error}"))
    })?;
    let mut value = String::with_capacity(33);
    value.push(prefix);
    for byte in bytes {
        use std::fmt::Write as _;
        write!(value, "{byte:02x}").expect("writing to String cannot fail");
    }
    OpaqueId::parse(value).map_err(durable_effect)
}

fn duration_seconds(duration: Duration) -> Result<f64, DeletionError> {
    let seconds = duration.as_secs_f64();
    if seconds.is_finite() && seconds > 0.0 {
        Ok(seconds)
    } else {
        Err(DeletionError::InvalidConfig("invalid claim ttl".into()))
    }
}

fn durable_effect(error: impl std::fmt::Display) -> DeletionError {
    DeletionError::Durable(remora_durable::DurableError::Effect(error.to_string()))
}

fn database_target_error(error: sqlx::Error) -> TargetError {
    TargetError::internal(format!("durable deletion state failed: {error}"))
}

#[derive(FromRow)]
struct JobRow {
    job_id: String,
    owner_id: String,
    phase: String,
    lifecycle: String,
    target_store: String,
    target_bucket: String,
    target_prefix: Value,
    committed_result: Option<Value>,
    deletion_operation_id: Option<String>,
}

#[derive(Deserialize)]
struct CommittedResult {
    attempt_id: OpaqueId,
    manifest_id: OpaqueId,
    schema_version: u32,
    manifest: CommittedManifest,
    member_count: u64,
    total_bytes: u64,
}

#[derive(Deserialize)]
struct CommittedManifest {
    store: String,
    bucket: String,
    key: String,
    version_id: Option<String>,
    sha256: String,
    size: u64,
}

#[derive(FromRow)]
struct InventoryValidationRow {
    state: String,
    manifest_schema_version: Option<i32>,
    manifest_id: Option<String>,
    manifest_store: Option<String>,
    manifest_bucket: Option<String>,
    manifest_key: Option<String>,
    manifest_version_id: Option<String>,
    manifest_sha256: Option<String>,
    manifest_size: Option<i64>,
    member_count: i64,
    total_bytes: i64,
    staging_origin_attempt_id: Option<String>,
    target_prefix: Option<String>,
    committed_attempt_job_id: Option<String>,
    staging_origin_job_id: Option<String>,
    actual_member_count: i64,
    actual_total_bytes: i64,
}

#[derive(FromRow)]
struct OwnedIdentityRow {
    kind: String,
    target_store: Option<String>,
    bucket: Option<String>,
    object_key: Option<String>,
    staging_origin_attempt_id: Option<String>,
    target_prefix: Option<String>,
    manifest_id: Option<String>,
    origin_job_id: Option<String>,
}

#[derive(FromRow)]
struct OperationRow {
    operation_id: String,
    job_id: String,
    state: String,
    problem_class: Option<String>,
    problem_message: Option<String>,
    created_at: OffsetDateTime,
    updated_at: OffsetDateTime,
    completed_at: Option<OffsetDateTime>,
}

impl OperationRow {
    fn into_operation(self) -> Result<DeletionOperation, DeletionError> {
        let problem = match (self.problem_class, self.problem_message) {
            (None, None) => None,
            (Some(class), Some(message)) => Some(DeletionProblem {
                class: parse_problem_class(&class)?,
                message,
            }),
            _ => return Err(DeletionError::CorruptInventory),
        };
        Ok(DeletionOperation {
            id: OpaqueId::parse(self.operation_id).map_err(durable_effect)?,
            job_id: OpaqueId::parse(self.job_id).map_err(durable_effect)?,
            state: parse_state(&self.state)?,
            problem,
            created_at: self.created_at,
            updated_at: self.updated_at,
            completed_at: self.completed_at,
        })
    }
}

#[derive(FromRow)]
struct OperationAccessRow {
    operation_id: String,
    job_id: String,
    state: String,
    problem_class: Option<String>,
    problem_message: Option<String>,
    created_at: OffsetDateTime,
    updated_at: OffsetDateTime,
    completed_at: Option<OffsetDateTime>,
    owner_id: String,
    target_store: String,
    target_bucket: String,
    target_prefix: Value,
}

impl OperationAccessRow {
    fn into_operation(self) -> Result<DeletionOperation, DeletionError> {
        OperationRow {
            operation_id: self.operation_id,
            job_id: self.job_id,
            state: self.state,
            problem_class: self.problem_class,
            problem_message: self.problem_message,
            created_at: self.created_at,
            updated_at: self.updated_at,
            completed_at: self.completed_at,
        }
        .into_operation()
    }
}

#[derive(FromRow)]
struct ClaimRow {
    item_id: i64,
    operation_id: String,
    kind: String,
    target_store: String,
    bucket: String,
    object_key: String,
    version_id: Option<String>,
    upload_id: Option<String>,
    claim_token: String,
}

impl ClaimRow {
    fn into_claim(self) -> Result<DeleteClaim, DeletionError> {
        let object = || ObjectIdentity {
            target_store_id: self.target_store.clone(),
            bucket: self.bucket.clone(),
            key: self.object_key.clone(),
            version_id: self.version_id.clone(),
        };
        let identity = match self.kind.as_str() {
            "write_intent" => DeleteIdentity::WriteIntent(object()),
            "object" => DeleteIdentity::Object(object()),
            "manifest" => DeleteIdentity::Manifest(object()),
            "multipart" => DeleteIdentity::Multipart(MultipartIdentity {
                target_store_id: self.target_store,
                bucket: self.bucket,
                key: self.object_key,
                upload_id: self.upload_id.ok_or(DeletionError::CorruptInventory)?,
            }),
            _ => return Err(DeletionError::CorruptInventory),
        };
        Ok(DeleteClaim {
            item_id: u64::try_from(self.item_id).map_err(|_| DeletionError::CorruptInventory)?,
            operation_id: OpaqueId::parse(self.operation_id).map_err(durable_effect)?,
            claim_token: OpaqueId::parse(self.claim_token).map_err(durable_effect)?,
            identity,
        })
    }
}