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 credentials;
mod deletions;
mod events;
mod job_store;
mod jobs;
mod projections;
mod request;
mod results;
mod routes;

use job_store::*;
use projections::*;
use request::*;

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

enum LifecycleAction {
    SoftDelete,
    Restore,
}

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

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

#[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 position(created_at: OffsetDateTime, id: &str) -> KeysetPosition {
    KeysetPosition {
        created_at,
        id: id.to_owned(),
    }
}

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

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