remora/crates/remora-deletion/src/service.rs

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,
};

mod inventory;
mod lifecycle;
mod operations;

use inventory::{
    finalize_if_empty, insert_discovered_multipart, insert_discovered_object, materialize_items,
    validate_committed_inventory, validate_deletion_identities,
};

#[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,
        })
    }
}

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 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()
}

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,
        })
    }
}