remora/crates/remora-api/src/public.rs

use std::{collections::HashSet, time::Duration};

use actix_web::{
    HttpRequest, HttpResponse, ResponseError,
    http::{StatusCode, header},
    web,
};
use remora_admission::{AdmissionDecision, AdmissionService};
use remora_attempt::{AttemptService, CommittedResultRecord, ProgressSnapshot};
use remora_contracts::{
    ApiErrorCodeDto, ApiErrorDto, CredentialMaterialDto, DateTimeDto, JobSubmissionDto,
    NullableDto, ResultMemberAccessDto, SourceCredentialCreateDto, SourceCredentialDto,
    SourceCredentialKindDto, SourceCredentialPageDto, SourceCredentialRotateDto,
    SourceCredentialStatusDto, SourceCredentialVersionDto,
};
use remora_credentials::{
    CreateCredential, CredentialCaller, CredentialError, CredentialRegistry, RotateTxOutcome,
    SecretMaterial, SourceCredential, SourceCredentialKind, SourceCredentialStatus,
};
use remora_deletion::{
    DeletionError, DeletionOperation, DeletionProblemClass, DeletionService, DeletionState,
    TargetRegistry,
};
use remora_domain::{
    BucketName, OpaqueId, Target, TargetPrefix, TargetStoreId, fingerprint, ports::MutationScope,
};
use remora_durable::{
    CursorCodec, DurableError, KeysetCursor, KeysetPosition, MutationDisposition, MutationStatus,
    PgTx, execute_mutation, first_page_cursor, next_page_cursor, observe_mutation,
};
use remora_identity::{
    AuthorizedResource, ManagementOperation, OidcIdentityProvider, Principal, may_manage,
    resolve_principal,
};
use remora_result::{CommittedResult, PageSize, PublicationInput, ResultInventory};
use remora_target::{ContentSha256, ObjectIdentity, StoredObject};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sqlx::{PgPool, types::Json};
use time::{OffsetDateTime, format_description::well_known::Rfc3339};

mod events;
mod routes;

const DEFAULT_LIMIT: u32 = 50;
const MAX_LIMIT: u32 = 200;

#[derive(Clone)]
pub struct PublicApiConfig {
    pub cursor_secret: Vec<u8>,
    pub member_access_ttl: Duration,
    pub event_retention: Duration,
    pub event_poll_interval: Duration,
    pub event_heartbeat_interval: Duration,
    pub event_ticket_ttl: Duration,
    pub event_allowed_origins: Vec<String>,
}

#[derive(Clone)]
pub struct PublicApiState {
    pool: PgPool,
    identity: OidcIdentityProvider,
    admission: AdmissionService,
    attempts: AttemptService,
    credentials: CredentialRegistry,
    deletion: DeletionService,
    targets: TargetRegistry,
    results: ResultInventory,
    cursors: CursorCodec,
    member_access_ttl: Duration,
    event_retention: Duration,
    event_poll_interval: Duration,
    event_heartbeat_interval: Duration,
    event_ticket_ttl: Duration,
    event_allowed_origins: HashSet<String>,
}

pub struct PublicApiServices {
    pub identity: OidcIdentityProvider,
    pub admission: AdmissionService,
    pub attempts: AttemptService,
    pub credentials: CredentialRegistry,
    pub deletion: DeletionService,
    pub targets: TargetRegistry,
}

impl PublicApiState {
    pub fn new(
        pool: PgPool,
        services: PublicApiServices,
        config: PublicApiConfig,
    ) -> Result<Self, DurableError> {
        if config.member_access_ttl.is_zero() {
            return Err(DurableError::Config(
                "member access expiry must be positive".into(),
            ));
        }
        if config.event_retention.is_zero()
            || config.event_poll_interval.is_zero()
            || config.event_heartbeat_interval.is_zero()
            || config.event_ticket_ttl.is_zero()
        {
            return Err(DurableError::Config(
                "event retention, intervals, and ticket expiry must be positive".into(),
            ));
        }
        if config.event_ticket_ttl > Duration::from_secs(60) {
            return Err(DurableError::Config(
                "event ticket expiry must not exceed 60 seconds".into(),
            ));
        }
        let event_allowed_origins = config
            .event_allowed_origins
            .into_iter()
            .collect::<HashSet<_>>();
        if event_allowed_origins.is_empty()
            || event_allowed_origins
                .iter()
                .any(|origin| origin.is_empty() || origin.trim() != origin)
        {
            return Err(DurableError::Config(
                "at least one exact event origin is required".into(),
            ));
        }
        Ok(Self {
            results: ResultInventory::new(pool.clone()),
            cursors: CursorCodec::new(config.cursor_secret)?,
            pool,
            identity: services.identity,
            admission: services.admission,
            attempts: services.attempts,
            credentials: services.credentials,
            deletion: services.deletion,
            targets: services.targets,
            member_access_ttl: config.member_access_ttl,
            event_retention: config.event_retention,
            event_poll_interval: config.event_poll_interval,
            event_heartbeat_interval: config.event_heartbeat_interval,
            event_ticket_ttl: config.event_ticket_ttl,
            event_allowed_origins,
        })
    }

    async fn principal(&self, request: &HttpRequest) -> Result<Principal, PublicApiError> {
        let authorization = request
            .headers()
            .get(header::AUTHORIZATION)
            .and_then(|value| value.to_str().ok())
            .ok_or_else(PublicApiError::unauthenticated)?;
        let external = self
            .identity
            .authenticate_token(authorization)
            .await
            .map_err(|error| match error {
                remora_identity::IdentityError::ProviderUnavailable(_) => {
                    PublicApiError::unavailable()
                }
                _ => PublicApiError::unauthenticated(),
            })?;
        resolve_principal(&self.pool, external)
            .await
            .map_err(|_| PublicApiError::unavailable())
    }
}

pub fn configure_public_api(config: &mut web::ServiceConfig) {
    config.service(
        web::scope("/v1")
            .app_data(
                web::JsonConfig::default()
                    .limit(1024 * 1024)
                    .error_handler(|_, _| PublicApiError::invalid_request().into()),
            )
            .configure(routes::jobs)
            .configure(routes::results)
            .configure(routes::credentials)
            .configure(routes::deletions)
            .configure(routes::events),
    );
}

pub async fn prune_live_event_history(
    pool: &PgPool,
    before: OffsetDateTime,
) -> Result<u64, DurableError> {
    events::prune_history(pool, before).await
}

#[derive(Deserialize)]
struct PageQuery {
    cursor: Option<String>,
    limit: Option<u32>,
}

#[derive(Deserialize, Serialize)]
struct JobPageQuery {
    cursor: Option<String>,
    limit: Option<u32>,
    #[serde(default)]
    include_deleted: bool,
}

async fn submit_job(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    request: web::Json<JobSubmissionDto>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let key = idempotency_key(&http)?;
    let request = request.into_inner();
    let request_fingerprint =
        fingerprint(&request).map_err(|_| PublicApiError::invalid_request())?;
    let admission = state.admission.clone();
    let state_for_projection = state.get_ref().clone();
    let principal_for_effect = principal.clone();
    let outcome = execute_mutation(
        &state.pool,
        MutationScope {
            subject: &principal.id,
            operation: "job.submit",
            key,
        },
        "submit_job",
        &[],
        request_fingerprint,
        move |tx| {
            Box::pin(async move {
                match admission
                    .submit_in_tx(tx, &principal_for_effect, request)
                    .await?
                {
                    AdmissionDecision::Accepted(accepted) => {
                        let projection = load_authorized_job_in_tx(
                            tx,
                            &state_for_projection,
                            &principal_for_effect,
                            &accepted.job_id,
                            false,
                        )
                        .await
                        .map_err(|error| DurableError::Effect(error.to_string()))?;
                        Ok((MutationStatus::Accepted, projection))
                    }
                    AdmissionDecision::Rejected { .. } => Ok((
                        MutationStatus::Rejected,
                        public_error_body(ApiErrorCodeDto::InvalidRequest, "invalid request"),
                    )),
                }
            })
        },
    )
    .await?;
    if outcome.disposition == MutationDisposition::Conflict {
        return Err(PublicApiError::idempotency_conflict());
    }
    if outcome.status == MutationStatus::Accepted {
        let job_id = outcome
            .response
            .get("id")
            .and_then(Value::as_str)
            .ok_or_else(PublicApiError::internal)?;
        let job_id = parse_id(job_id.to_owned())?;
        authorize_job(&state, &principal, &job_id, ManagementOperation::Read).await?;
    }
    let response = if outcome.status == MutationStatus::Rejected {
        HttpResponse::BadRequest()
            .content_type("application/problem+json")
            .json(outcome.response)
    } else {
        HttpResponse::Accepted().json(outcome.response)
    };
    Ok(with_idempotency_expiry(response))
}

async fn get_job(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    path: web::Path<String>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let job_id = parse_id(path.into_inner())?;
    let job = load_authorized_job(&state, &principal, &job_id, true).await?;
    Ok(HttpResponse::Ok().json(job))
}

async fn list_jobs(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    query: web::Query<JobPageQuery>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let limit = page_limit(query.limit)?;
    let filters = json!({
        "resource": "jobs",
        "include_deleted": query.include_deleted,
        "authorization": authorization_snapshot(&principal)
    });
    let filter = CursorCodec::filter_fingerprint(&filters)?;
    let cursor = query
        .cursor
        .as_deref()
        .map(|token| state.cursors.bind(token, principal.id.as_str(), filter))
        .transpose()
        .map_err(PublicApiError::from_cursor)?;
    let event_cursor = event_cursor(&state, &principal).await?;
    let mut rows = load_job_page_rows(
        &state.pool,
        &principal,
        query.include_deleted,
        cursor.as_ref(),
        limit + 1,
    )
    .await?;
    let has_more = rows.len() > limit as usize;
    rows.truncate(limit as usize);
    let mut items = Vec::with_capacity(rows.len());
    for row in &rows {
        items.push(job_row_projection(
            row,
            current_progress(&state, row).await?,
            &event_cursor,
        )?);
    }
    let next_cursor = if has_more {
        let last = rows.last().ok_or_else(PublicApiError::internal)?;
        let last = position(last.created_at, &last.job_id);
        let cursor = match cursor.as_ref() {
            Some(current) => next_page_cursor(current, last),
            None => first_page_cursor(
                principal.id.as_str(),
                filter,
                position(rows[0].created_at, &rows[0].job_id),
                last,
            ),
        };
        Some(state.cursors.encode(&cursor)?)
    } else {
        None
    };
    Ok(HttpResponse::Ok().json(json!({
        "items": items,
        "next_cursor": next_cursor,
        "event_cursor": event_cursor,
        "management_capabilities": {
            "revoke_execution_grants": principal.administrator
        }
    })))
}

async fn get_job_result(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    path: web::Path<String>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let job_id = parse_id(path.into_inner())?;
    let row = load_visible_job_row(&state, &principal, &job_id).await?;
    let result = committed_result(&row)?.ok_or_else(PublicApiError::precondition_failed)?;
    state.results.resolve_committed(&result).await?;
    Ok(HttpResponse::Ok().json(result_projection(&result)))
}

async fn list_result_members(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    path: web::Path<String>,
    query: web::Query<PageQuery>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let job_id = parse_id(path.into_inner())?;
    let row = load_visible_job_row(&state, &principal, &job_id).await?;
    let committed = committed_result(&row)?.ok_or_else(PublicApiError::precondition_failed)?;
    let filter = CursorCodec::filter_fingerprint(&json!({
        "resource": "result_members",
        "job_id": job_id,
        "authorization": authorization_snapshot(&principal)
    }))?;
    let cursor = query
        .cursor
        .as_deref()
        .map(|token| state.cursors.bind(token, principal.id.as_str(), filter))
        .transpose()
        .map_err(PublicApiError::from_cursor)?;
    let after = cursor
        .as_ref()
        .and_then(|value| value.after.as_ref())
        .map(|value| value.id.as_str());
    let page_size = PageSize::new(page_limit(query.limit)?)?;
    let page = state
        .results
        .page_members(&committed, after, page_size)
        .await?;
    let next_cursor = if let Some(after) = page.next_after.as_deref() {
        let next = KeysetPosition {
            created_at: row.created_at,
            id: after.to_owned(),
        };
        let cursor = match cursor.as_ref() {
            Some(current) => next_page_cursor(current, next),
            None => first_page_cursor(
                principal.id.as_str(),
                filter,
                KeysetPosition {
                    created_at: row.created_at,
                    id: String::new(),
                },
                next,
            ),
        };
        Some(state.cursors.encode(&cursor)?)
    } else {
        None
    };
    let items = page
        .items
        .iter()
        .map(|member| {
            json!({
                "id": member.id(),
                "path": member.path(),
                "size": member.size(),
                "object": object_identity_projection(
                    member.object().store(),
                    member.object().bucket(),
                    member.object().key(),
                    member.object().version_id(),
                    member.content_sha256().as_str(),
                    member.size()
                )
            })
        })
        .collect::<Vec<_>>();
    Ok(HttpResponse::Ok().json(json!({
        "items": items,
        "next_cursor": next_cursor
    })))
}

async fn access_result_member(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    path: web::Path<(String, String)>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let (job_id, member_id) = path.into_inner();
    let job_id = parse_id(job_id)?;
    let member_id = parse_id(member_id)?;
    let row = load_visible_job_row(&state, &principal, &job_id).await?;
    let committed = committed_result(&row)?.ok_or_else(PublicApiError::precondition_failed)?;
    let member = state.results.lookup_member(&committed, &member_id).await?;
    let target = state
        .targets
        .resolve(member.object().store(), member.object().bucket())
        .ok_or_else(PublicApiError::unavailable)?;
    let issued_at = OffsetDateTime::now_utc();
    let access = target
        .presign_read(&member.exact_identity(), state.member_access_ttl)
        .await
        .map_err(|_| PublicApiError::unavailable())?;
    let duration =
        time::Duration::try_from(access.expires_in).map_err(|_| PublicApiError::internal())?;
    Ok(HttpResponse::Ok()
        .insert_header((header::CACHE_CONTROL, "no-store"))
        .insert_header((header::PRAGMA, "no-cache"))
        .json(ResultMemberAccessDto {
            url: access.url,
            expires_at: datetime(issued_at + duration)?,
        }))
}

async fn cancel_job(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    path: web::Path<String>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let job_id = parse_id(path.into_inner())?;
    authorize_job(&state, &principal, &job_id, ManagementOperation::Cancel).await?;
    let key = idempotency_key(&http)?.to_owned();
    let job = idempotent_job_mutation(
        &state,
        &principal,
        &job_id,
        &key,
        "job.cancel",
        JobMutationAction::Cancel,
    )
    .await?;
    Ok(with_idempotency_expiry(HttpResponse::Accepted().json(job)))
}

async fn soft_delete_job(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    path: web::Path<String>,
) -> Result<HttpResponse, PublicApiError> {
    lifecycle_mutation(state, http, path, LifecycleAction::SoftDelete).await
}

async fn restore_job(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    path: web::Path<String>,
) -> Result<HttpResponse, PublicApiError> {
    lifecycle_mutation(state, http, path, LifecycleAction::Restore).await
}

enum LifecycleAction {
    SoftDelete,
    Restore,
}

async fn lifecycle_mutation(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    path: web::Path<String>,
    action: LifecycleAction,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let job_id = parse_id(path.into_inner())?;
    let key = idempotency_key(&http)?.to_owned();
    let (operation, management) = match action {
        LifecycleAction::SoftDelete => ("job.soft_delete", ManagementOperation::SoftDelete),
        LifecycleAction::Restore => ("job.restore", ManagementOperation::Restore),
    };
    authorize_job(&state, &principal, &job_id, management).await?;
    let action = match action {
        LifecycleAction::SoftDelete => JobMutationAction::SoftDelete,
        LifecycleAction::Restore => JobMutationAction::Restore,
    };
    let job = idempotent_job_mutation(&state, &principal, &job_id, &key, operation, action).await?;
    Ok(with_idempotency_expiry(HttpResponse::Ok().json(job)))
}

async fn hard_delete_job(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    path: web::Path<String>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let job_id = parse_id(path.into_inner())?;
    authorize_job(&state, &principal, &job_id, ManagementOperation::HardDelete).await?;
    let key = idempotency_key(&http)?.to_owned();
    let request_fingerprint =
        fingerprint(&json!({ "job_id": job_id })).map_err(|_| PublicApiError::internal())?;
    let principal_for_effect = principal.clone();
    let job_for_effect = job_id.clone();
    let outcome = execute_mutation(
        &state.pool,
        MutationScope {
            subject: &principal.id,
            operation: "job.hard_delete",
            key: &key,
        },
        "hard_delete_job",
        &[],
        request_fingerprint,
        move |tx| {
            Box::pin(async move {
                let outcome =
                    DeletionService::hard_delete_in_tx(tx, &principal_for_effect, &job_for_effect)
                        .await
                        .map_err(DurableError::effect)?;
                Ok((
                    MutationStatus::Accepted,
                    deletion_projection(&outcome.operation)
                        .map_err(|error| DurableError::Effect(error.to_string()))?,
                ))
            })
        },
    )
    .await?;
    if outcome.disposition == MutationDisposition::Conflict {
        return Err(PublicApiError::idempotency_conflict());
    }
    Ok(with_idempotency_expiry(
        HttpResponse::Accepted().json(outcome.response),
    ))
}

async fn revoke_job(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    path: web::Path<String>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    if !principal.administrator {
        return Err(PublicApiError::forbidden());
    }
    let job_id = parse_id(path.into_inner())?;
    let row = load_job_row(&state.pool, &job_id)
        .await?
        .ok_or_else(PublicApiError::not_found)?;
    let key = idempotency_key(&http)?.to_owned();
    let grant_id = parse_id(row.execution_grant_id.clone())?;
    let job = idempotent_job_mutation(
        &state,
        &principal,
        &job_id,
        &key,
        "job.revoke",
        JobMutationAction::Revoke { grant_id },
    )
    .await?;
    Ok(with_idempotency_expiry(HttpResponse::Accepted().json(job)))
}

async fn list_job_attempts(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    path: web::Path<String>,
    query: web::Query<PageQuery>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let job_id = parse_id(path.into_inner())?;
    authorize_job(&state, &principal, &job_id, ManagementOperation::Read).await?;
    let limit = page_limit(query.limit)?;
    let filter = CursorCodec::filter_fingerprint(&json!({
        "resource": "attempts",
        "job_id": job_id,
        "authorization": authorization_snapshot(&principal)
    }))?;
    let cursor = query
        .cursor
        .as_deref()
        .map(|token| state.cursors.bind(token, principal.id.as_str(), filter))
        .transpose()
        .map_err(PublicApiError::from_cursor)?;
    let after = cursor
        .as_ref()
        .and_then(|value| value.after.as_ref())
        .map(|value| value.id.parse::<i64>())
        .transpose()
        .map_err(|_| PublicApiError::cursor_invalid())?;
    let rows = sqlx::query_as::<_, AttemptRow>(
        "SELECT a.attempt_id, a.fence, a.worker_incarnation, a.state, a.failure,
                CASE WHEN j.committed_result->>'attempt_id' = a.attempt_id
                     THEN j.committed_result END AS result,
                a.authorized_at,
                COALESCE(a.committed_at, a.stopped_at, a.reaped_at, a.executing_at, a.authorized_at)
                    AS updated_at
         FROM attempts a
         JOIN jobs j ON j.job_id = a.job_id
         WHERE a.job_id = $1 AND ($2::bigint IS NULL OR a.fence < $2)
         ORDER BY a.fence DESC
         LIMIT $3",
    )
    .bind(job_id.as_str())
    .bind(after)
    .bind(i64::from(limit) + 1)
    .fetch_all(&state.pool)
    .await?;
    let has_more = rows.len() > limit as usize;
    let visible = rows.iter().take(limit as usize).collect::<Vec<_>>();
    let progress = state
        .attempts
        .progress_snapshot(&job_id)
        .await
        .unwrap_or(None);
    let items = visible
        .iter()
        .map(|row| attempt_projection(row, progress.as_ref()))
        .collect::<Result<Vec<_>, _>>()?;
    let next_cursor = if has_more {
        let last = visible.last().ok_or_else(PublicApiError::internal)?;
        let next = KeysetPosition {
            created_at: last.authorized_at,
            id: last.fence.to_string(),
        };
        let cursor = match cursor.as_ref() {
            Some(current) => next_page_cursor(current, next),
            None => first_page_cursor(
                principal.id.as_str(),
                filter,
                KeysetPosition {
                    created_at: rows[0].authorized_at,
                    id: rows[0].fence.to_string(),
                },
                next,
            ),
        };
        Some(state.cursors.encode(&cursor)?)
    } else {
        None
    };
    Ok(HttpResponse::Ok().json(json!({
        "items": items,
        "next_cursor": next_cursor
    })))
}

async fn create_source_credential(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    request: web::Json<SourceCredentialCreateDto>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let key = idempotency_key(&http)?;
    let request = request.into_inner();
    let request_fingerprint =
        fingerprint(&request).map_err(|_| PublicApiError::invalid_request())?;
    let create = create_credential(request)?;
    let credentials = state.credentials.clone();
    let owner = principal.id.clone();
    let scope = MutationScope {
        subject: &principal.id,
        operation: "credential.create",
        key,
    };
    if let Some(outcome) = observe_mutation(
        &state.pool,
        scope,
        "create_source_credential",
        &[],
        request_fingerprint,
    )
    .await?
    {
        return credential_mutation_response(outcome, StatusCode::CREATED);
    }
    let prepared = credentials
        .prepare_create(&owner, create)
        .await
        .map_err(PublicApiError::from_credential)?;
    let outcome = execute_mutation(
        &state.pool,
        scope,
        "create_source_credential",
        &[],
        request_fingerprint,
        move |tx| {
            Box::pin(async move {
                let credential = CredentialRegistry::commit_prepared_create_in_tx(tx, prepared)
                    .await
                    .map_err(|error| DurableError::Effect(error.to_string()))?;
                let response = serde_json::to_value(
                    credential_dto(credential)
                        .map_err(|error| DurableError::Effect(error.to_string()))?,
                )
                .map_err(|error| DurableError::Effect(error.to_string()))?;
                Ok((MutationStatus::Accepted, response))
            })
        },
    )
    .await?;
    credential_mutation_response(outcome, StatusCode::CREATED)
}

async fn list_source_credentials(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    query: web::Query<PageQuery>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let caller = credential_caller(&principal);
    let filter = CursorCodec::filter_fingerprint(&json!({
        "resource": "credentials",
        "authorization": authorization_snapshot(&principal)
    }))?;
    let cursor = query
        .cursor
        .as_deref()
        .map(|token| state.cursors.bind(token, principal.id.as_str(), filter))
        .transpose()
        .map_err(PublicApiError::from_cursor)?;
    let registry_cursor = cursor
        .as_ref()
        .and_then(|value| value.after.as_ref())
        .map(|value| value.id.as_str());
    let page = state
        .credentials
        .list(&caller, page_limit(query.limit)?, registry_cursor)
        .await
        .map_err(PublicApiError::from_credential)?;
    let next_cursor = page
        .next_cursor
        .as_deref()
        .map(|raw| {
            let next = KeysetPosition {
                created_at: OffsetDateTime::UNIX_EPOCH,
                id: raw.to_owned(),
            };
            let bound = match cursor.as_ref() {
                Some(current) => next_page_cursor(current, next),
                None => first_page_cursor(
                    principal.id.as_str(),
                    filter,
                    KeysetPosition {
                        created_at: OffsetDateTime::UNIX_EPOCH,
                        id: String::new(),
                    },
                    next,
                ),
            };
            state.cursors.encode(&bound)
        })
        .transpose()?;
    let dto = SourceCredentialPageDto {
        items: page
            .items
            .into_iter()
            .map(credential_dto)
            .collect::<Result<_, _>>()?,
        next_cursor: NullableDto(next_cursor),
    };
    Ok(HttpResponse::Ok().json(dto))
}

async fn get_source_credential(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    path: web::Path<String>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let credential_id = parse_id(path.into_inner())?;
    let credential = state
        .credentials
        .get(&credential_caller(&principal), &credential_id)
        .await
        .map_err(PublicApiError::from_credential)?
        .ok_or_else(PublicApiError::not_found)?;
    Ok(HttpResponse::Ok().json(credential_dto(credential)?))
}

async fn rotate_source_credential(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    path: web::Path<String>,
    request: web::Json<SourceCredentialRotateDto>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let credential_id = parse_id(path.into_inner())?;
    let key = idempotency_key(&http)?;
    let request = request.into_inner();
    let request_fingerprint = fingerprint(&json!({
        "credential_id": credential_id,
        "request": request
    }))
    .map_err(|_| PublicApiError::invalid_request())?;
    let credentials = state.credentials.clone();
    let caller = credential_caller(&principal);
    let scope = MutationScope {
        subject: &principal.id,
        operation: "credential.rotate",
        key,
    };
    if let Some(outcome) = observe_mutation(
        &state.pool,
        scope,
        "rotate_source_credential",
        &[],
        request_fingerprint,
    )
    .await?
    {
        return credential_mutation_response(outcome, StatusCode::CREATED);
    }
    let existing = credentials
        .get(&caller, &credential_id)
        .await
        .map_err(PublicApiError::from_credential)?
        .ok_or_else(PublicApiError::not_found)?;
    let material = credential_material_for_kind(request.material, existing.kind);
    let prepared = credentials
        .prepare_rotate(&caller, &credential_id, material)
        .await
        .map_err(PublicApiError::from_credential)?;
    let outcome = execute_mutation(
        &state.pool,
        scope,
        "rotate_source_credential",
        &[],
        request_fingerprint,
        move |tx| {
            Box::pin(async move {
                match CredentialRegistry::commit_prepared_rotate_in_tx(tx, prepared).await? {
                    RotateTxOutcome::Applied(credential) => {
                        let response = serde_json::to_value(
                            credential_dto(credential)
                                .map_err(|error| DurableError::Effect(error.to_string()))?,
                        )
                        .map_err(|error| DurableError::Effect(error.to_string()))?;
                        Ok((MutationStatus::Accepted, response))
                    }
                    RotateTxOutcome::Revoked => Ok((
                        MutationStatus::Rejected,
                        public_error_body(
                            ApiErrorCodeDto::PreconditionFailed,
                            "precondition failed",
                        ),
                    )),
                    RotateTxOutcome::IntentClaimed => Err(DurableError::Effect(
                        "credential rotation intent is unavailable".into(),
                    )),
                }
            })
        },
    )
    .await?;
    credential_mutation_response(outcome, StatusCode::CREATED)
}

async fn revoke_source_credential(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    path: web::Path<String>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let credential_id = parse_id(path.into_inner())?;
    let key = idempotency_key(&http)?;
    let request_fingerprint = fingerprint(&json!({ "credential_id": credential_id }))
        .map_err(|_| PublicApiError::internal())?;
    let caller = credential_caller(&principal);
    let scope = MutationScope {
        subject: &principal.id,
        operation: "credential.revoke",
        key,
    };
    if let Some(outcome) = observe_mutation(
        &state.pool,
        scope,
        "revoke_source_credential",
        &[],
        request_fingerprint,
    )
    .await?
    {
        return credential_mutation_response(outcome, StatusCode::OK);
    }
    state
        .credentials
        .get(&caller, &credential_id)
        .await
        .map_err(PublicApiError::from_credential)?
        .ok_or_else(PublicApiError::not_found)?;
    let credential_for_effect = credential_id.clone();
    let outcome = execute_mutation(
        &state.pool,
        scope,
        "revoke_source_credential",
        &[],
        request_fingerprint,
        move |tx| {
            Box::pin(async move {
                let credential =
                    CredentialRegistry::revoke_in_tx(tx, &caller, &credential_for_effect)
                        .await
                        .map_err(|error| DurableError::Effect(error.to_string()))?;
                let response = serde_json::to_value(
                    credential_dto(credential)
                        .map_err(|error| DurableError::Effect(error.to_string()))?,
                )
                .map_err(|error| DurableError::Effect(error.to_string()))?;
                Ok((MutationStatus::Accepted, response))
            })
        },
    )
    .await?;
    credential_mutation_response(outcome, StatusCode::OK)
}

async fn get_deletion_operation(
    state: web::Data<PublicApiState>,
    http: HttpRequest,
    path: web::Path<String>,
) -> Result<HttpResponse, PublicApiError> {
    let principal = state.principal(&http).await?;
    let operation_id = parse_id(path.into_inner())?;
    let operation = state
        .deletion
        .operation(&principal, &operation_id)
        .await
        .map_err(PublicApiError::from_deletion)?;
    Ok(HttpResponse::Ok().json(deletion_projection(&operation)?))
}

enum JobMutationAction {
    Cancel,
    SoftDelete,
    Restore,
    Revoke { grant_id: OpaqueId },
}

async fn idempotent_job_mutation(
    state: &PublicApiState,
    principal: &Principal,
    job_id: &OpaqueId,
    key: &str,
    operation: &'static str,
    action: JobMutationAction,
) -> Result<Value, PublicApiError> {
    let request_fingerprint =
        fingerprint(&json!({ "job_id": job_id })).map_err(|_| PublicApiError::internal())?;
    let state_for_projection = state.clone();
    let principal_for_projection = principal.clone();
    let job_for_projection = job_id.clone();
    let outcome = execute_mutation(
        &state.pool,
        MutationScope {
            subject: &principal.id,
            operation,
            key,
        },
        operation,
        &[],
        request_fingerprint,
        move |tx| {
            Box::pin(async move {
                match action {
                    JobMutationAction::Cancel => {
                        AttemptService::request_cancel_in_tx(tx, &job_for_projection).await?;
                    }
                    JobMutationAction::SoftDelete => {
                        DeletionService::soft_delete_in_tx(
                            tx,
                            &principal_for_projection,
                            &job_for_projection,
                        )
                        .await
                        .map_err(DurableError::effect)?;
                    }
                    JobMutationAction::Restore => {
                        DeletionService::restore_in_tx(
                            tx,
                            &principal_for_projection,
                            &job_for_projection,
                        )
                        .await
                        .map_err(DurableError::effect)?;
                    }
                    JobMutationAction::Revoke { grant_id } => {
                        remora_admission::AdmissionService::revoke_execution_grant_in_tx(
                            tx,
                            &principal_for_projection,
                            &grant_id,
                        )
                        .await?;
                        AttemptService::request_cancel_in_tx(tx, &job_for_projection).await?;
                    }
                }
                let projection = load_authorized_job_in_tx(
                    tx,
                    &state_for_projection,
                    &principal_for_projection,
                    &job_for_projection,
                    true,
                )
                .await
                .map_err(|error| DurableError::Effect(error.to_string()))?;
                Ok((MutationStatus::Accepted, projection))
            })
        },
    )
    .await?;
    if outcome.disposition == MutationDisposition::Conflict {
        Err(PublicApiError::idempotency_conflict())
    } else {
        Ok(outcome.response)
    }
}

#[derive(sqlx::FromRow)]
struct JobRow {
    job_id: String,
    owner_id: String,
    label: Option<String>,
    revision: i64,
    phase: String,
    lifecycle: String,
    source_projection: Json<Value>,
    target_store: String,
    target_bucket: String,
    target_prefix: Json<Vec<String>>,
    execution_grant_id: String,
    authority_attempt_id: Option<String>,
    authority_worker_incarnation: Option<String>,
    current_fence: i64,
    checkpoint: Option<Json<Value>>,
    committed_result: Option<Json<Value>>,
    failure: Option<Json<Value>>,
    deletion_operation_id: Option<String>,
    hard_delete_started_at: Option<OffsetDateTime>,
    created_at: OffsetDateTime,
    updated_at: OffsetDateTime,
}

async fn load_job_row(pool: &PgPool, job_id: &OpaqueId) -> Result<Option<JobRow>, PublicApiError> {
    Ok(sqlx::query_as::<_, JobRow>(
        "SELECT job_id, owner_id, label, revision, phase, lifecycle, source_projection,
                target_store, target_bucket, target_prefix, execution_grant_id,
                authority_attempt_id,
                (SELECT worker_incarnation FROM attempts WHERE attempt_id = jobs.authority_attempt_id)
                    AS authority_worker_incarnation,
                current_fence, checkpoint, committed_result, failure,
                deletion_operation_id, hard_delete_started_at, created_at, updated_at
         FROM jobs WHERE job_id = $1",
    )
    .bind(job_id.as_str())
    .fetch_optional(pool)
    .await?)
}

async fn load_job_row_in_tx(
    tx: &mut PgTx<'_>,
    job_id: &OpaqueId,
) -> Result<Option<JobRow>, PublicApiError> {
    Ok(sqlx::query_as::<_, JobRow>(
        "SELECT job_id, owner_id, label, revision, phase, lifecycle, source_projection,
                target_store, target_bucket, target_prefix, execution_grant_id,
                authority_attempt_id,
                (SELECT worker_incarnation FROM attempts WHERE attempt_id = jobs.authority_attempt_id)
                    AS authority_worker_incarnation,
                current_fence, checkpoint, committed_result, failure,
                deletion_operation_id, hard_delete_started_at, created_at, updated_at
         FROM jobs WHERE job_id = $1",
    )
    .bind(job_id.as_str())
    .fetch_optional(tx.as_mut())
    .await?)
}

async fn load_authorized_job_in_tx(
    tx: &mut PgTx<'_>,
    state: &PublicApiState,
    principal: &Principal,
    job_id: &OpaqueId,
    include_deleted: bool,
) -> Result<Value, PublicApiError> {
    let row = load_job_row_in_tx(tx, job_id)
        .await?
        .ok_or_else(PublicApiError::not_found)?;
    if !include_deleted && row.lifecycle != "visible" {
        return Err(PublicApiError::not_found());
    }
    authorize_row(principal, &row, ManagementOperation::Read)?;
    let cursor = event_cursor_in_tx(tx, state, principal).await?;
    job_row_projection(&row, None, &cursor)
}

async fn load_authorized_job(
    state: &PublicApiState,
    principal: &Principal,
    job_id: &OpaqueId,
    include_deleted: bool,
) -> Result<Value, PublicApiError> {
    let cursor = event_cursor(state, principal).await?;
    let row = load_job_row(&state.pool, job_id)
        .await?
        .ok_or_else(PublicApiError::not_found)?;
    if !include_deleted && row.lifecycle != "visible" {
        return Err(PublicApiError::not_found());
    }
    authorize_row(principal, &row, ManagementOperation::Read)?;
    let progress = current_progress(state, &row).await?;
    job_row_projection(&row, progress, &cursor)
}

async fn load_visible_job_row(
    state: &PublicApiState,
    principal: &Principal,
    job_id: &OpaqueId,
) -> Result<JobRow, PublicApiError> {
    let row = load_job_row(&state.pool, job_id)
        .await?
        .ok_or_else(PublicApiError::not_found)?;
    if row.lifecycle != "visible" {
        return Err(PublicApiError::not_found());
    }
    authorize_row(principal, &row, ManagementOperation::Read)?;
    Ok(row)
}

async fn authorize_job(
    state: &PublicApiState,
    principal: &Principal,
    job_id: &OpaqueId,
    operation: ManagementOperation,
) -> Result<JobRow, PublicApiError> {
    let row = load_job_row(&state.pool, job_id)
        .await?
        .ok_or_else(PublicApiError::not_found)?;
    authorize_row(principal, &row, operation)?;
    Ok(row)
}

fn authorize_row(
    principal: &Principal,
    row: &JobRow,
    operation: ManagementOperation,
) -> Result<(), PublicApiError> {
    let resource = AuthorizedResource {
        owner: parse_id(row.owner_id.clone())?,
        target: row_target(row)?,
    };
    if may_manage(principal, &resource, operation) {
        Ok(())
    } else {
        Err(PublicApiError::not_found())
    }
}

async fn load_job_page_rows(
    pool: &PgPool,
    principal: &Principal,
    include_deleted: bool,
    cursor: Option<&KeysetCursor>,
    wanted: u32,
) -> Result<Vec<JobRow>, PublicApiError> {
    let mut after = cursor.and_then(|value| value.after.clone());
    let high_water = cursor.map(|value| value.high_water.clone());
    let mut visible = Vec::new();
    loop {
        let rows = sqlx::query_as::<_, JobRow>(
            "SELECT job_id, owner_id, label, revision, phase, lifecycle, source_projection,
                    target_store, target_bucket, target_prefix, execution_grant_id,
                    authority_attempt_id,
                    (SELECT worker_incarnation FROM attempts WHERE attempt_id = jobs.authority_attempt_id)
                        AS authority_worker_incarnation,
                    current_fence, checkpoint, committed_result, failure,
                    deletion_operation_id, hard_delete_started_at, created_at, updated_at
             FROM jobs
             WHERE ($1::text IS NULL OR owner_id = $1)
               AND ($2 OR lifecycle = 'visible')
               AND ($3::timestamptz IS NULL OR (created_at, job_id) <= ($3, $4))
               AND ($5::timestamptz IS NULL OR (created_at, job_id) < ($5, $6))
             ORDER BY created_at DESC, job_id DESC
             LIMIT 200",
        )
        .bind((!principal.administrator).then(|| principal.id.as_str()))
        .bind(include_deleted)
        .bind(high_water.as_ref().map(|value| value.created_at))
        .bind(high_water.as_ref().map(|value| value.id.as_str()))
        .bind(after.as_ref().map(|value| value.created_at))
        .bind(after.as_ref().map(|value| value.id.as_str()))
        .fetch_all(pool)
        .await?;
        if rows.is_empty() {
            break;
        }
        let count = rows.len();
        for row in rows {
            after = Some(position(row.created_at, &row.job_id));
            if authorize_row(principal, &row, ManagementOperation::List).is_ok() {
                visible.push(row);
                if visible.len() >= wanted as usize {
                    return Ok(visible);
                }
            }
        }
        if count < 200 {
            break;
        }
    }
    Ok(visible)
}

async fn current_progress(
    state: &PublicApiState,
    row: &JobRow,
) -> Result<Option<ProgressSnapshot>, PublicApiError> {
    let Some(authority_attempt_id) = row.authority_attempt_id.as_deref() else {
        return Ok(None);
    };
    let Ok(progress) = state
        .attempts
        .progress_snapshot(&parse_id(row.job_id.clone())?)
        .await
    else {
        return Ok(None);
    };
    Ok(progress.filter(|progress| {
        progress.attempt_id.as_str() == authority_attempt_id
            && i64::try_from(progress.fence).ok() == Some(row.current_fence)
    }))
}

fn progress_projection(progress: &ProgressSnapshot) -> Result<Value, PublicApiError> {
    Ok(json!({
        "attempt_id": progress.attempt_id,
        "fence": progress.fence,
        "completed_bytes": progress.progress.completed_bytes,
        "total_bytes": progress.progress.total_bytes,
        "completed_members": progress.progress.completed_members,
        "total_members": progress.progress.total_members,
        "observed_at": format_time(progress.observed_at)?
    }))
}

fn job_row_projection(
    row: &JobRow,
    progress: Option<ProgressSnapshot>,
    event_cursor: &str,
) -> Result<Value, PublicApiError> {
    let target = json!({
        "store": row.target_store,
        "bucket": row.target_bucket,
        "prefix": row.target_prefix.0.join("/")
    });
    if row.lifecycle == "hard_deleted" || row.lifecycle == "hard_deleting" {
        return Ok(json!({
            "resource_kind": "tombstone",
            "id": row.job_id,
            "owner": row.owner_id,
            "target": target,
            "terminal_phase": phase_projection(&row.phase)?,
            "deletion_operation_id": row.deletion_operation_id,
            "created_at": format_time(row.created_at)?,
            "hard_delete_started_at": row.hard_delete_started_at.map(format_time).transpose()?,
            "updated_at": format_time(row.updated_at)?,
            "event_cursor": event_cursor
        }));
    }
    let authority = row
        .authority_attempt_id
        .as_ref()
        .zip(row.authority_worker_incarnation.as_ref())
        .map(|(attempt_id, worker_incarnation)| {
            json!({
                "attempt_id": attempt_id,
                "fence": row.current_fence,
                "worker_incarnation": worker_incarnation
            })
        });
    let checkpoint = row
        .checkpoint
        .as_ref()
        .map(|value| checkpoint_projection(&value.0))
        .transpose()?;
    let result = committed_result(row)?.as_ref().map(result_projection);
    let failure = row
        .failure
        .as_ref()
        .map(|value| failure_projection(&value.0))
        .transpose()?;
    Ok(json!({
        "resource_kind": "job",
        "id": row.job_id,
        "owner": row.owner_id,
        "label": row.label,
        "revision": row.revision,
        "phase": phase_projection(&row.phase)?,
        "lifecycle": row.lifecycle,
        "source": row.source_projection.0,
        "target": target,
        "authoritative_attempt": authority,
        "progress": progress.as_ref().map(progress_projection).transpose()?,
        "checkpoint": checkpoint,
        "result": result,
        "failure": failure,
        "created_at": format_time(row.created_at)?,
        "updated_at": format_time(row.updated_at)?,
        "event_cursor": event_cursor
    }))
}

fn phase_projection(value: &str) -> Result<&'static str, PublicApiError> {
    match value {
        "queued" => Ok("Queued"),
        "running" => Ok("Running"),
        "retry_wait" => Ok("RetryWait"),
        "cancelling" => Ok("Cancelling"),
        "succeeded" => Ok("Succeeded"),
        "failed" => Ok("Failed"),
        "cancelled" => Ok("Cancelled"),
        _ => Err(PublicApiError::internal()),
    }
}

fn checkpoint_projection(value: &Value) -> Result<Value, PublicApiError> {
    let state = value
        .get("state")
        .and_then(Value::as_str)
        .ok_or_else(PublicApiError::internal)?;
    match state {
        "reusable" => {
            let checkpoint = value
                .get("checkpoint")
                .ok_or_else(PublicApiError::internal)?;
            Ok(json!({
                "reusable": true,
                "validated_bytes": checkpoint.get("validated_bytes"),
                "validated_members": checkpoint.get("validated_members")
            }))
        }
        "complete" => Ok(json!({
            "reusable": false,
            "validated_bytes": value.get("validated_bytes"),
            "validated_members": value.get("validated_members")
        })),
        _ => Err(PublicApiError::internal()),
    }
}

fn failure_projection(value: &Value) -> Result<Value, PublicApiError> {
    let class = value
        .get("class")
        .and_then(Value::as_str)
        .ok_or_else(PublicApiError::internal)?;
    let disposition = value
        .get("disposition")
        .and_then(Value::as_str)
        .ok_or_else(PublicApiError::internal)?;
    Ok(json!({
        "class": class,
        "message": "job failed",
        "retryable": disposition == "retryable"
    }))
}

fn row_target(row: &JobRow) -> Result<Target, PublicApiError> {
    Ok(Target {
        store: TargetStoreId::parse(row.target_store.clone())
            .map_err(|_| PublicApiError::internal())?,
        bucket: BucketName::parse(row.target_bucket.clone())
            .map_err(|_| PublicApiError::internal())?,
        prefix: TargetPrefix::parse(&row.target_prefix.0.join("/"), std::iter::empty())
            .map_err(|_| PublicApiError::internal())?,
    })
}

fn committed_result(row: &JobRow) -> Result<Option<PublicationInput>, PublicApiError> {
    let Some(value) = row.committed_result.as_ref() else {
        return Ok(None);
    };
    let record: CommittedResultRecord =
        serde_json::from_value(value.0.clone()).map_err(|_| PublicApiError::internal())?;
    Ok(Some(PublicationInput {
        attempt_id: record.attempt_id,
        manifest_id: record.manifest_id,
        schema_version: record.schema_version,
        manifest: StoredObject {
            identity: ObjectIdentity {
                target_store_id: record.manifest.store,
                bucket: record.manifest.bucket,
                key: record.manifest.key,
                version_id: record.manifest.version_id,
            },
            content_sha256: ContentSha256::parse(record.manifest.sha256)
                .map_err(|_| PublicApiError::internal())?,
            size: record.manifest.size,
        },
        member_count: record.member_count,
        total_bytes: record.total_bytes,
    }))
}

fn result_projection(result: &CommittedResult) -> Value {
    json!({
        "kind": "manifest",
        "manifest_id": result.manifest_id,
        "schema_version": result.schema_version,
        "manifest": object_identity_projection(
            &result.manifest.identity.target_store_id,
            &result.manifest.identity.bucket,
            &result.manifest.identity.key,
            result.manifest.identity.version_id.as_deref(),
            result.manifest.content_sha256.as_str(),
            result.manifest.size
        ),
        "member_count": result.member_count,
        "total_bytes": result.total_bytes
    })
}

fn object_identity_projection(
    store: &str,
    bucket: &str,
    key: &str,
    version_id: Option<&str>,
    checksum: &str,
    size: u64,
) -> Value {
    let mut object = json!({
        "store": store,
        "bucket": bucket,
        "key": key,
        "checksum": checksum,
        "size": size
    });
    if let Some(version_id) = version_id {
        object["version_id"] = json!(version_id);
    }
    object
}

#[derive(sqlx::FromRow)]
struct AttemptRow {
    attempt_id: String,
    fence: i64,
    worker_incarnation: String,
    state: String,
    failure: Option<Json<Value>>,
    result: Option<Json<Value>>,
    authorized_at: OffsetDateTime,
    updated_at: OffsetDateTime,
}

fn attempt_projection(
    row: &AttemptRow,
    progress: Option<&ProgressSnapshot>,
) -> Result<Value, PublicApiError> {
    let state = match row.state.as_str() {
        "authorized" => "AttemptAuthorized",
        "executing" => "AttemptExecuting",
        "reaped" | "stopped" => "AttemptStopped",
        "committed" => "AttemptCommitted",
        _ => return Err(PublicApiError::internal()),
    };
    let result = row
        .result
        .as_ref()
        .map(|value| result_record_projection(&value.0))
        .transpose()?;
    let progress = progress
        .filter(|progress| {
            progress.attempt_id.as_str() == row.attempt_id
                && i64::try_from(progress.fence).ok() == Some(row.fence)
        })
        .map(progress_projection)
        .transpose()?;
    Ok(json!({
        "id": row.attempt_id,
        "fence": row.fence,
        "worker_incarnation": row.worker_incarnation,
        "state": state,
        "progress": progress,
        "result": result,
        "failure": row
            .failure
            .as_ref()
            .map(|value| failure_projection(&value.0))
            .transpose()?,
        "created_at": format_time(row.authorized_at)?,
        "updated_at": format_time(row.updated_at)?
    }))
}

fn result_record_projection(value: &Value) -> Result<Value, PublicApiError> {
    let record: CommittedResultRecord =
        serde_json::from_value(value.clone()).map_err(|_| PublicApiError::internal())?;
    Ok(json!({
        "kind": "manifest",
        "manifest_id": record.manifest_id,
        "schema_version": record.schema_version,
        "manifest": object_identity_projection(
            &record.manifest.store,
            &record.manifest.bucket,
            &record.manifest.key,
            record.manifest.version_id.as_deref(),
            &record.manifest.sha256,
            record.manifest.size
        ),
        "member_count": record.member_count,
        "total_bytes": record.total_bytes
    }))
}

fn create_credential(
    request: SourceCredentialCreateDto,
) -> Result<CreateCredential, PublicApiError> {
    let (label, material) = match request {
        SourceCredentialCreateDto::BasicAuth(value) => (
            value.label,
            SecretMaterial::BasicAuth {
                username: value.material.username,
                password: value.material.password,
            },
        ),
        SourceCredentialCreateDto::BearerToken(value) => (
            value.label,
            SecretMaterial::BearerToken {
                token: value.material.token,
            },
        ),
        SourceCredentialCreateDto::HttpHeaders(value) => (
            value.label,
            SecretMaterial::HttpHeaders {
                headers: value.material.headers,
            },
        ),
        SourceCredentialCreateDto::ObjectStore(value) => (
            value.label,
            SecretMaterial::ObjectStore {
                access_key_id: value.material.access_key_id,
                secret_access_key: value.material.secret_access_key,
                session_token: value.material.session_token,
            },
        ),
        SourceCredentialCreateDto::HuggingFace(value) => (
            value.label,
            SecretMaterial::HuggingFace {
                token: value.material.token,
            },
        ),
    };
    Ok(CreateCredential { label, material })
}

fn credential_material(material: CredentialMaterialDto) -> SecretMaterial {
    match material {
        CredentialMaterialDto::BasicAuth(value) => SecretMaterial::BasicAuth {
            username: value.username,
            password: value.password,
        },
        CredentialMaterialDto::Token(value) => SecretMaterial::BearerToken { token: value.token },
        CredentialMaterialDto::HttpHeaders(value) => SecretMaterial::HttpHeaders {
            headers: value.headers,
        },
        CredentialMaterialDto::ObjectStore(value) => SecretMaterial::ObjectStore {
            access_key_id: value.access_key_id,
            secret_access_key: value.secret_access_key,
            session_token: value.session_token,
        },
    }
}

fn credential_material_for_kind(
    material: CredentialMaterialDto,
    kind: SourceCredentialKind,
) -> SecretMaterial {
    match (material, kind) {
        (CredentialMaterialDto::Token(value), SourceCredentialKind::HuggingFace) => {
            SecretMaterial::HuggingFace { token: value.token }
        }
        (material, _) => credential_material(material),
    }
}

fn credential_mutation_response(
    outcome: remora_durable::MutationOutcome,
    accepted_status: StatusCode,
) -> Result<HttpResponse, PublicApiError> {
    if outcome.disposition == MutationDisposition::Conflict {
        return Err(PublicApiError::idempotency_conflict());
    }
    let response = if outcome.status == MutationStatus::Rejected {
        HttpResponse::PreconditionFailed()
            .content_type("application/problem+json")
            .json(outcome.response)
    } else {
        HttpResponse::build(accepted_status).json(outcome.response)
    };
    Ok(with_idempotency_expiry(response))
}

fn public_error_body(code: ApiErrorCodeDto, message: &'static str) -> Value {
    json!({
        "code": code,
        "message": message,
        "request_id": request_id()
    })
}

fn credential_dto(value: SourceCredential) -> Result<SourceCredentialDto, PublicApiError> {
    let kind = match value.kind {
        SourceCredentialKind::BasicAuth => SourceCredentialKindDto::BasicAuth,
        SourceCredentialKind::BearerToken => SourceCredentialKindDto::BearerToken,
        SourceCredentialKind::HttpHeaders => SourceCredentialKindDto::HttpHeaders,
        SourceCredentialKind::ObjectStore => SourceCredentialKindDto::ObjectStore,
        SourceCredentialKind::HuggingFace => SourceCredentialKindDto::HuggingFace,
    };
    let status = match value.status {
        SourceCredentialStatus::Active => SourceCredentialStatusDto::Active,
        SourceCredentialStatus::Revoked => SourceCredentialStatusDto::Revoked,
    };
    Ok(SourceCredentialDto {
        id: value.id,
        owner: value.owner,
        label: value.label,
        kind,
        status,
        active_version: NullableDto(
            value
                .active_version
                .map(|version| {
                    Ok::<_, PublicApiError>(SourceCredentialVersionDto {
                        id: version.id,
                        created_at: datetime(version.created_at)?,
                    })
                })
                .transpose()?,
        ),
        created_at: datetime(value.created_at)?,
        updated_at: datetime(value.updated_at)?,
    })
}

fn credential_caller(principal: &Principal) -> CredentialCaller {
    CredentialCaller {
        principal_id: principal.id.clone(),
        administrator: principal.administrator,
    }
}

fn deletion_projection(operation: &DeletionOperation) -> Result<Value, PublicApiError> {
    let state = match operation.state {
        DeletionState::Queued => "queued",
        DeletionState::Running => "running",
        DeletionState::RetryWait => "retry_wait",
        DeletionState::Blocked => "blocked",
        DeletionState::Completed => "completed",
    };
    let problem = operation.problem.as_ref().map(|problem| {
        let class = match problem.class {
            DeletionProblemClass::Authorization => "authorization",
            DeletionProblemClass::TargetUnavailable => "target_unavailable",
            DeletionProblemClass::Configuration => "configuration",
            DeletionProblemClass::Internal => "internal",
        };
        json!({ "class": class, "message": problem.message })
    });
    Ok(json!({
        "id": operation.id,
        "job_id": operation.job_id,
        "state": state,
        "problem": problem,
        "created_at": format_time(operation.created_at)?,
        "updated_at": format_time(operation.updated_at)?
    }))
}

async fn event_cursor(
    state: &PublicApiState,
    principal: &Principal,
) -> Result<String, PublicApiError> {
    let high_water = sqlx::query_as::<_, (i64, OffsetDateTime)>(
        "SELECT GREATEST(
                    COALESCE((SELECT MAX(id) FROM outbox_messages), 0),
                    history.floor_outbox_id
                ),
                now()
         FROM live_event_history_state history
         WHERE history.singleton",
    )
    .fetch_one(&state.pool)
    .await?;
    encode_event_cursor(state, principal, high_water)
}

async fn event_cursor_in_tx(
    tx: &mut PgTx<'_>,
    state: &PublicApiState,
    principal: &Principal,
) -> Result<String, PublicApiError> {
    let high_water = sqlx::query_as::<_, (i64, OffsetDateTime)>(
        "SELECT GREATEST(
                    COALESCE((SELECT MAX(id) FROM outbox_messages), 0),
                    history.floor_outbox_id
                ),
                now()
         FROM live_event_history_state history
         WHERE history.singleton",
    )
    .fetch_one(tx.as_mut())
    .await?;
    encode_event_cursor(state, principal, high_water)
}

fn encode_event_cursor(
    state: &PublicApiState,
    principal: &Principal,
    high_water: (i64, OffsetDateTime),
) -> Result<String, PublicApiError> {
    let created_at = high_water.1;
    let id = format!("event-{}", high_water.0);
    let filter = CursorCodec::filter_fingerprint(&json!({
        "resource": "events",
        "authorization": authorization_snapshot(principal)
    }))?;
    state
        .cursors
        .encode(&first_page_cursor(
            principal.id.as_str(),
            filter,
            KeysetPosition {
                created_at,
                id: id.clone(),
            },
            KeysetPosition { created_at, id },
        ))
        .map_err(Into::into)
}

fn position(created_at: OffsetDateTime, id: &str) -> KeysetPosition {
    KeysetPosition {
        created_at,
        id: id.to_owned(),
    }
}

fn authorization_snapshot(principal: &Principal) -> Value {
    let mut scopes = principal
        .scopes
        .iter()
        .map(|scope| {
            (
                scope.store.as_str(),
                scope.bucket.as_str(),
                scope.prefix.canonical(),
            )
        })
        .collect::<Vec<_>>();
    scopes.sort_unstable();
    scopes.dedup();
    json!({
        "administrator": principal.administrator,
        "scopes": scopes
    })
}

fn page_limit(limit: Option<u32>) -> Result<u32, PublicApiError> {
    let limit = limit.unwrap_or(DEFAULT_LIMIT);
    if !(1..=MAX_LIMIT).contains(&limit) {
        return Err(PublicApiError::invalid_request());
    }
    Ok(limit)
}

fn parse_id(value: impl Into<String>) -> Result<OpaqueId, PublicApiError> {
    OpaqueId::parse(value).map_err(|_| PublicApiError::invalid_request())
}

fn idempotency_key(request: &HttpRequest) -> Result<&str, PublicApiError> {
    let value = request
        .headers()
        .get("Idempotency-Key")
        .and_then(|value| value.to_str().ok())
        .ok_or_else(PublicApiError::invalid_request)?;
    if !(16..=128).contains(&value.len())
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || b"._~-".contains(&byte))
    {
        return Err(PublicApiError::invalid_request());
    }
    Ok(value)
}

fn with_idempotency_expiry(mut response: HttpResponse) -> HttpResponse {
    let expiry = OffsetDateTime::now_utc() + time::Duration::hours(24);
    if let Ok(value) = expiry.format(&Rfc3339)
        && let Ok(value) = header::HeaderValue::from_str(&value)
    {
        response.headers_mut().insert(
            header::HeaderName::from_static("idempotency-expires-at"),
            value,
        );
    }
    response
}

fn format_time(value: OffsetDateTime) -> Result<String, PublicApiError> {
    value
        .format(&Rfc3339)
        .map_err(|_| PublicApiError::internal())
}

fn datetime(value: OffsetDateTime) -> Result<DateTimeDto, PublicApiError> {
    DateTimeDto::parse(format_time(value)?).map_err(|_| PublicApiError::internal())
}

#[derive(Debug)]
struct PublicApiError {
    status: StatusCode,
    code: ApiErrorCodeDto,
    message: &'static str,
}

impl PublicApiError {
    fn invalid_request() -> Self {
        Self::new(
            StatusCode::BAD_REQUEST,
            ApiErrorCodeDto::InvalidRequest,
            "invalid request",
        )
    }

    fn unauthenticated() -> Self {
        Self::new(
            StatusCode::UNAUTHORIZED,
            ApiErrorCodeDto::Unauthenticated,
            "authentication required",
        )
    }

    fn forbidden() -> Self {
        Self::new(
            StatusCode::FORBIDDEN,
            ApiErrorCodeDto::Forbidden,
            "operation forbidden",
        )
    }

    fn not_found() -> Self {
        Self::new(
            StatusCode::NOT_FOUND,
            ApiErrorCodeDto::NotFound,
            "resource not found",
        )
    }

    fn idempotency_conflict() -> Self {
        Self::new(
            StatusCode::CONFLICT,
            ApiErrorCodeDto::IdempotencyConflict,
            "idempotency key conflict",
        )
    }

    fn precondition_failed() -> Self {
        Self::new(
            StatusCode::CONFLICT,
            ApiErrorCodeDto::PreconditionFailed,
            "resource precondition failed",
        )
    }

    fn cursor_invalid() -> Self {
        Self::new(
            StatusCode::BAD_REQUEST,
            ApiErrorCodeDto::CursorInvalid,
            "invalid cursor",
        )
    }

    fn unavailable() -> Self {
        Self::new(
            StatusCode::SERVICE_UNAVAILABLE,
            ApiErrorCodeDto::Unavailable,
            "service unavailable",
        )
    }

    fn internal() -> Self {
        Self::new(
            StatusCode::INTERNAL_SERVER_ERROR,
            ApiErrorCodeDto::Internal,
            "internal error",
        )
    }

    fn new(status: StatusCode, code: ApiErrorCodeDto, message: &'static str) -> Self {
        Self {
            status,
            code,
            message,
        }
    }

    fn from_cursor(error: DurableError) -> Self {
        match error {
            DurableError::InvalidCursor | DurableError::CursorBindingMismatch => {
                Self::cursor_invalid()
            }
            _ => Self::unavailable(),
        }
    }

    fn from_credential(error: CredentialError) -> Self {
        match error {
            CredentialError::NotFound => Self::not_found(),
            CredentialError::InvalidInput(_) | CredentialError::KindMismatch => {
                Self::invalid_request()
            }
            CredentialError::Revoked | CredentialError::NoActiveVersion => {
                Self::precondition_failed()
            }
            CredentialError::SecretStoreUnavailable => Self::unavailable(),
            _ => Self::unavailable(),
        }
    }

    fn from_deletion(error: DeletionError) -> Self {
        match error {
            DeletionError::NotFound => Self::not_found(),
            DeletionError::NotTerminal | DeletionError::HardDeleteStarted => {
                Self::precondition_failed()
            }
            DeletionError::CorruptInventory => Self::precondition_failed(),
            _ => Self::unavailable(),
        }
    }
}

impl std::fmt::Display for PublicApiError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(self.message)
    }
}

impl ResponseError for PublicApiError {
    fn status_code(&self) -> StatusCode {
        self.status
    }

    fn error_response(&self) -> HttpResponse {
        HttpResponse::build(self.status)
            .content_type("application/problem+json")
            .json(ApiErrorDto {
                code: self.code,
                message: self.message.to_owned(),
                request_id: request_id(),
                fields: None,
            })
    }
}

impl From<sqlx::Error> for PublicApiError {
    fn from(_: sqlx::Error) -> Self {
        Self::unavailable()
    }
}

impl From<DurableError> for PublicApiError {
    fn from(error: DurableError) -> Self {
        Self::from_cursor(error)
    }
}

impl From<remora_result::ResultError> for PublicApiError {
    fn from(error: remora_result::ResultError) -> Self {
        match error {
            remora_result::ResultError::MemberNotFound => Self::not_found(),
            remora_result::ResultError::NotSealed => Self::precondition_failed(),
            _ => Self::unavailable(),
        }
    }
}

fn request_id() -> OpaqueId {
    let mut bytes = [0_u8; 16];
    getrandom::fill(&mut bytes).expect("operating system randomness is required");
    let mut value = String::from("r");
    for byte in bytes {
        use std::fmt::Write as _;
        write!(value, "{byte:02x}").expect("writing to String cannot fail");
    }
    OpaqueId::parse(value).expect("random request id is valid")
}

#[cfg(test)]
mod tests {
    use remora_domain::StorageScope;

    use super::*;

    #[test]
    fn cursor_authorization_snapshot_is_order_independent_and_change_sensitive() {
        let scope = |prefix: &str| StorageScope {
            store: TargetStoreId::parse("target").unwrap(),
            bucket: BucketName::parse("bucket").unwrap(),
            prefix: TargetPrefix::parse(prefix, std::iter::empty()).unwrap(),
        };
        let principal = |administrator, scopes| Principal {
            id: OpaqueId::parse("principal").unwrap(),
            issuer: "https://issuer.example".into(),
            subject: "subject".into(),
            administrator,
            scopes,
        };
        let first = principal(false, vec![scope("b"), scope("a")]);
        let reordered = principal(false, vec![scope("a"), scope("b")]);
        let duplicated = principal(false, vec![scope("a"), scope("b"), scope("a")]);
        let expanded = principal(false, vec![scope("a"), scope("b"), scope("c")]);
        let administrator = principal(true, vec![scope("a"), scope("b")]);

        assert_eq!(
            authorization_snapshot(&first),
            authorization_snapshot(&reordered)
        );
        assert_eq!(
            authorization_snapshot(&first),
            authorization_snapshot(&duplicated)
        );
        assert_ne!(
            authorization_snapshot(&first),
            authorization_snapshot(&expanded)
        );
        assert_ne!(
            authorization_snapshot(&reordered),
            authorization_snapshot(&administrator)
        );

        let codec = CursorCodec::new(b"public-api-test-cursor-secret".to_vec()).unwrap();
        let filter = CursorCodec::filter_fingerprint(&authorization_snapshot(&first)).unwrap();
        let position = KeysetPosition {
            created_at: OffsetDateTime::UNIX_EPOCH,
            id: "job".into(),
        };
        let token = codec
            .encode(&first_page_cursor(
                first.id.as_str(),
                filter,
                position.clone(),
                position,
            ))
            .unwrap();
        assert!(codec.bind(&token, "different-subject", filter).is_err());
        let expanded_filter =
            CursorCodec::filter_fingerprint(&authorization_snapshot(&expanded)).unwrap();
        assert!(
            codec
                .bind(&token, first.id.as_str(), expanded_filter)
                .is_err()
        );
    }
}