lodestone-client 0.0.16

//! Recovery-aware Collection portability workflows over the generated API.

use std::error::Error;
use std::fmt::{self, Display, Formatter};

use crate::{CallOptions, Client, ClientError, OperationWaitError, OperationWaitOptions, SubmissionDisposition};
use lodestone_wire::{
    CollectionExportArtifact, Digest, Operation, OperationState, PortabilityStorageCapability,
    PrepareCollectionExportRequest, PrepareCollectionRestoreRequest, ResolveCollectionExportRequest,
    ResolveCollectionRestoreRequest, StartCollectionExportRequest, StartCollectionRestoreRequest,
};
use serde::Serialize;

pub const PORTABILITY_WORKFLOW_FORMAT_VERSION: u16 = 1;
const PORTABILITY_WAIT_ATTEMPTS: u32 = 32;
const PORTABILITY_WAIT_INTERVAL_MILLIS: u64 = 250;

#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct CollectionExportExecutionV1 {
    pub operation: Operation,
    pub artifact: Option<CollectionExportArtifact>,
}

#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct CollectionRestoreExecutionV1 {
    pub operation: Operation,
}

#[derive(Debug)]
pub enum PortabilityWorkflowError {
    Client(ClientError),
    ExpiredUnknown,
    MissingOperation,
    MissingArtifact,
    Wait(OperationWaitError),
}

impl Display for PortabilityWorkflowError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::Client(_) => "Collection portability API call failed",
            Self::ExpiredUnknown => "Collection portability outcome retired before resolution",
            Self::MissingOperation => "Collection portability resolver omitted the operation",
            Self::MissingArtifact => "successful Collection export omitted its artifact",
            Self::Wait(_) => "Collection portability operation wait failed",
        })
    }
}

impl Error for PortabilityWorkflowError {
    fn source(&self) -> Option<&(dyn Error + 'static)> {
        match self {
            Self::Client(error) => Some(error),
            Self::Wait(error) => Some(error),
            Self::ExpiredUnknown | Self::MissingOperation | Self::MissingArtifact => None,
        }
    }
}

impl From<ClientError> for PortabilityWorkflowError {
    fn from(error: ClientError) -> Self {
        Self::Client(error)
    }
}

fn resolved_operation(
    operation: Option<Operation>,
    expired_unknown: bool,
) -> Result<Operation, PortabilityWorkflowError> {
    operation.ok_or(if expired_unknown {
        PortabilityWorkflowError::ExpiredUnknown
    } else {
        PortabilityWorkflowError::MissingOperation
    })
}

impl Client {
    /// Prepares, submits, waits once through the bounded native waiter, and
    /// resolves a Collection export. An indeterminate start response is never
    /// replayed blindly; the caller action identity is resolved instead.
    pub async fn export_collection(
        &self,
        action_id: Digest,
        storage: PortabilityStorageCapability,
        options: CallOptions,
    ) -> Result<CollectionExportExecutionV1, PortabilityWorkflowError> {
        let preparation = self
            .prepare_collection_export(
                PrepareCollectionExportRequest {
                    action_id: action_id.clone(),
                    storage: storage.clone(),
                },
                options.clone(),
            )
            .await?;
        let start = self
            .start_collection_export(
                StartCollectionExportRequest {
                    action_id: action_id.clone(),
                    semantic_digest: preparation.semantic_digest.clone(),
                    storage: storage.clone(),
                },
                options.clone(),
            )
            .await;
        if let Err(error) = start
            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
        {
            return Err(PortabilityWorkflowError::Client(error));
        }
        let request = ResolveCollectionExportRequest {
            action_id,
            semantic_digest: preparation.semantic_digest,
            storage,
        };
        let mut resolved = self.resolve_collection_export(request.clone(), options.clone()).await?;
        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
        if operation.current_state == OperationState::Pending {
            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
            resolved = self.resolve_collection_export(request, options).await?;
            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
        }
        if operation.current_state == OperationState::Succeeded && resolved.artifact.is_none() {
            return Err(PortabilityWorkflowError::MissingArtifact);
        }
        Ok(CollectionExportExecutionV1 {
            operation,
            artifact: resolved.artifact,
        })
    }

    /// Prepares, submits, waits once through the bounded native waiter, and
    /// resolves a restore into the caller's fresh destination Collection.
    pub async fn restore_collection(
        &self,
        action_id: Digest,
        artifact: CollectionExportArtifact,
        storage: PortabilityStorageCapability,
        options: CallOptions,
    ) -> Result<CollectionRestoreExecutionV1, PortabilityWorkflowError> {
        let preparation = self
            .prepare_collection_restore(
                PrepareCollectionRestoreRequest {
                    action_id: action_id.clone(),
                    artifact: artifact.clone(),
                    storage: storage.clone(),
                },
                options.clone(),
            )
            .await?;
        let start = self
            .start_collection_restore(
                StartCollectionRestoreRequest {
                    action_id: action_id.clone(),
                    semantic_digest: preparation.semantic_digest.clone(),
                    artifact: artifact.clone(),
                    storage: storage.clone(),
                },
                options.clone(),
            )
            .await;
        if let Err(error) = start
            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
        {
            return Err(PortabilityWorkflowError::Client(error));
        }
        let request = ResolveCollectionRestoreRequest {
            action_id,
            semantic_digest: preparation.semantic_digest,
            artifact,
            storage,
        };
        let mut resolved = self
            .resolve_collection_restore(request.clone(), options.clone())
            .await?;
        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
        if operation.current_state == OperationState::Pending {
            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
            resolved = self.resolve_collection_restore(request, options).await?;
            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
        }
        Ok(CollectionRestoreExecutionV1 { operation })
    }

    async fn wait_portability_operation(
        &self,
        operation: &Operation,
        options: CallOptions,
    ) -> Result<Operation, PortabilityWorkflowError> {
        Ok(self
            .wait_for_operation(
                operation.locator.clone(),
                operation.authority_revision.clone(),
                OperationWaitOptions {
                    maximum_attempts: PORTABILITY_WAIT_ATTEMPTS,
                    interval_millis: PORTABILITY_WAIT_INTERVAL_MILLIS,
                },
                options,
            )
            .await
            .map_err(PortabilityWorkflowError::Wait)?
            .operation)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn portability_wait_contract_is_bounded() {
        assert!((1..=64).contains(&PORTABILITY_WAIT_ATTEMPTS));
        assert!((1..=60_000).contains(&PORTABILITY_WAIT_INTERVAL_MILLIS));
        assert_eq!(PORTABILITY_WORKFLOW_FORMAT_VERSION, 1);
    }
}