lodestone-client 0.0.29

//! Recovery-aware 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, CollectionExportRetirement, CollectionTransferArtifact, Digest, IndexTransferArtifact,
    Operation, OperationState, PortabilityStorageCapability, PrepareCollectionExportRequest,
    PrepareCollectionExportRetirementRequest, PrepareCollectionRestoreRequest, PrepareCollectionTransferExportRequest,
    PrepareCollectionTransferRestoreRequest, PrepareIndexExportRequest, PrepareIndexImportRequest,
    ResolveCollectionExportRequest, ResolveCollectionExportRetirementRequest, ResolveCollectionRestoreRequest,
    ResolveCollectionTransferExportRequest, ResolveCollectionTransferRestoreRequest, ResolveIndexExportRequest,
    ResolveIndexImportRequest, Resource, StartCollectionExportRequest, StartCollectionExportRetirementRequest,
    StartCollectionRestoreRequest, StartCollectionTransferExportRequest, StartCollectionTransferRestoreRequest,
    StartIndexExportRequest, StartIndexImportRequest,
};
use serde::Serialize;

pub const PORTABILITY_WORKFLOW_FORMAT_VERSION: u16 = 4;
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(Clone, Debug, PartialEq, Serialize)]
pub struct CollectionTransferExportExecutionV1 {
    pub operation: Operation,
    pub artifact: Option<CollectionTransferArtifact>,
}

#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct CollectionTransferRestoreExecutionV1 {
    pub operation: Operation,
    pub verified_content_digest: Option<Digest>,
}

#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct CollectionExportRetirementExecutionV1 {
    pub operation: Operation,
    pub retirement: Option<CollectionExportRetirement>,
}

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

#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct IndexImportExecutionV1 {
    pub operation: Operation,
    pub destination: Option<Resource>,
}

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

impl Display for PortabilityWorkflowError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::Client(_) => "portability API call failed",
            Self::ExpiredUnknown => "portability outcome retired before resolution",
            Self::MissingOperation => "portability resolver omitted the operation",
            Self::MissingArtifact => "successful portability export omitted its artifact",
            Self::MissingRetirement => "successful Collection export retirement omitted its absence proof",
            Self::MissingDestination => "successful index import omitted its destination",
            Self::MissingVerification => "successful Collection transfer restore omitted its content verification",
            Self::VerificationMismatch => {
                "successful Collection transfer restore returned another content verification"
            }
            Self::Wait(_) => "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
            | Self::MissingRetirement
            | Self::MissingDestination
            | Self::MissingVerification
            | Self::VerificationMismatch => 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,
        })
    }

    /// Retires one exact Collection export and resolves the authoritative
    /// absence proof. A pending resolver outcome is followed by one exact
    /// replay of the same action identity and semantic digest; no fresh
    /// deletion identity is invented.
    ///
    /// Callers that require recovery across their own process loss must durably
    /// retain the prepare result and complete request before invoking start,
    /// then use the generated resolve/start methods with those exact values.
    pub async fn retire_collection_export(
        &self,
        action_id: Digest,
        artifact: CollectionExportArtifact,
        storage: PortabilityStorageCapability,
        options: CallOptions,
    ) -> Result<CollectionExportRetirementExecutionV1, PortabilityWorkflowError> {
        let preparation = self
            .prepare_collection_export_retirement(
                PrepareCollectionExportRetirementRequest {
                    action_id: action_id.clone(),
                    artifact: artifact.clone(),
                    storage: storage.clone(),
                },
                options.clone(),
            )
            .await?;
        let start_request = StartCollectionExportRetirementRequest {
            action_id: action_id.clone(),
            semantic_digest: preparation.semantic_digest.clone(),
            artifact: artifact.clone(),
            storage: storage.clone(),
        };
        let start = self
            .start_collection_export_retirement(start_request.clone(), options.clone())
            .await;
        if let Err(error) = start
            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
        {
            return Err(PortabilityWorkflowError::Client(error));
        }
        let resolve_request = ResolveCollectionExportRetirementRequest {
            action_id,
            semantic_digest: preparation.semantic_digest,
            artifact,
            storage,
        };
        let mut resolved = self
            .resolve_collection_export_retirement(resolve_request.clone(), options.clone())
            .await?;
        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
        if operation.current_state == OperationState::Pending {
            // Recovery is permitted only after the read-only resolver proves
            // that this exact retained operation is still pending.
            let _replayed = self
                .start_collection_export_retirement(start_request, options.clone())
                .await;
            resolved = self
                .resolve_collection_export_retirement(resolve_request.clone(), options.clone())
                .await?;
            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_retirement(resolve_request, options)
                .await?;
            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
        }
        if operation.current_state == OperationState::Succeeded && resolved.retirement.is_none() {
            return Err(PortabilityWorkflowError::MissingRetirement);
        }
        Ok(CollectionExportRetirementExecutionV1 {
            operation,
            retirement: resolved.retirement,
        })
    }

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

    /// Captures one bounded live Collection through the Core-owned admission
    /// fence and resolves its immutable base-plus-delta artifact. The complete
    /// prepare request is the caller's durable recovery identity.
    pub async fn export_collection_transfer(
        &self,
        request: PrepareCollectionTransferExportRequest,
        options: CallOptions,
    ) -> Result<CollectionTransferExportExecutionV1, PortabilityWorkflowError> {
        let preparation = self
            .prepare_collection_transfer_export(request.clone(), options.clone())
            .await?;
        let start = self
            .start_collection_transfer_export(
                StartCollectionTransferExportRequest {
                    action_id: request.action_id.clone(),
                    semantic_digest: preparation.semantic_digest.clone(),
                    expected_security_generation: request.expected_security_generation.clone(),
                    storage: request.storage.clone(),
                },
                options.clone(),
            )
            .await;
        if let Err(error) = start
            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
        {
            return Err(PortabilityWorkflowError::Client(error));
        }
        let resolve_request = ResolveCollectionTransferExportRequest {
            action_id: request.action_id,
            semantic_digest: preparation.semantic_digest,
            expected_security_generation: request.expected_security_generation,
            storage: request.storage,
        };
        let mut resolved = self
            .resolve_collection_transfer_export(resolve_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_transfer_export(resolve_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(CollectionTransferExportExecutionV1 {
            operation,
            artifact: resolved.artifact,
        })
    }

    /// Restores and verifies one exact fenced Collection transfer artifact.
    /// Submitted ambiguity is resolved only with the original action,
    /// artifact and storage capability.
    pub async fn restore_collection_transfer(
        &self,
        action_id: Digest,
        artifact: CollectionTransferArtifact,
        storage: PortabilityStorageCapability,
        options: CallOptions,
    ) -> Result<CollectionTransferRestoreExecutionV1, PortabilityWorkflowError> {
        let preparation = self
            .prepare_collection_transfer_restore(
                PrepareCollectionTransferRestoreRequest {
                    action_id: action_id.clone(),
                    artifact: artifact.clone(),
                    storage: storage.clone(),
                },
                options.clone(),
            )
            .await?;
        let expected_final_content_digest = preparation.final_content_digest.clone();
        let start = self
            .start_collection_transfer_restore(
                StartCollectionTransferRestoreRequest {
                    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 resolve_request = ResolveCollectionTransferRestoreRequest {
            action_id,
            semantic_digest: preparation.semantic_digest,
            artifact,
            storage,
        };
        let mut resolved = self
            .resolve_collection_transfer_restore(resolve_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_transfer_restore(resolve_request, options)
                .await?;
            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
        }
        if operation.current_state == OperationState::Succeeded && resolved.verified_content_digest.is_none() {
            return Err(PortabilityWorkflowError::MissingVerification);
        }
        if operation.current_state == OperationState::Succeeded
            && resolved.verified_content_digest.as_ref() != Some(&expected_final_content_digest)
        {
            return Err(PortabilityWorkflowError::VerificationMismatch);
        }
        Ok(CollectionTransferRestoreExecutionV1 {
            operation,
            verified_content_digest: resolved.verified_content_digest,
        })
    }

    /// Exports one exact source index into bounded, independently verifiable
    /// transfer pages. The original resource identity and caller-selected
    /// bounds remain part of every recovery request.
    pub async fn export_index(
        &self,
        request: PrepareIndexExportRequest,
        options: CallOptions,
    ) -> Result<IndexExportExecutionV1, PortabilityWorkflowError> {
        let preparation = self.prepare_index_export(request.clone(), options.clone()).await?;
        let start = self
            .start_index_export(
                StartIndexExportRequest {
                    action_id: request.action_id.clone(),
                    semantic_digest: preparation.semantic_digest.clone(),
                    index: request.index.clone(),
                    storage: request.storage.clone(),
                    maximum_records: request.maximum_records,
                    maximum_pages: request.maximum_pages,
                    maximum_page_bytes: request.maximum_page_bytes.clone(),
                },
                options.clone(),
            )
            .await;
        if let Err(error) = start
            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
        {
            return Err(PortabilityWorkflowError::Client(error));
        }
        let resolve_request = ResolveIndexExportRequest {
            action_id: request.action_id,
            semantic_digest: preparation.semantic_digest,
            index: request.index,
            storage: request.storage,
            maximum_records: request.maximum_records,
            maximum_pages: request.maximum_pages,
            maximum_page_bytes: request.maximum_page_bytes,
        };
        let mut resolved = self
            .resolve_index_export(resolve_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_index_export(resolve_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(IndexExportExecutionV1 {
            operation,
            artifact: resolved.artifact,
        })
    }

    /// Imports one verified transfer artifact into a fresh destination name.
    /// Destination policy, topology, and storage choices are caller-owned and
    /// remain bound to the same recovery identity.
    pub async fn import_index(
        &self,
        request: PrepareIndexImportRequest,
        options: CallOptions,
    ) -> Result<IndexImportExecutionV1, PortabilityWorkflowError> {
        let preparation = self.prepare_index_import(request.clone(), options.clone()).await?;
        let start = self
            .start_index_import(
                StartIndexImportRequest {
                    action_id: request.action_id.clone(),
                    semantic_digest: preparation.semantic_digest.clone(),
                    artifact: request.artifact.clone(),
                    storage: request.storage.clone(),
                    destination_name: request.destination_name.clone(),
                    expected_name_absent: request.expected_name_absent,
                    destination_policies: request.destination_policies.clone(),
                    destination_resource_spec: request.destination_resource_spec.clone(),
                },
                options.clone(),
            )
            .await;
        if let Err(error) = start
            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
        {
            return Err(PortabilityWorkflowError::Client(error));
        }
        let resolve_request = ResolveIndexImportRequest {
            action_id: request.action_id,
            semantic_digest: preparation.semantic_digest,
            artifact: request.artifact,
            storage: request.storage,
            destination_name: request.destination_name,
            expected_name_absent: request.expected_name_absent,
            destination_policies: request.destination_policies,
            destination_resource_spec: request.destination_resource_spec,
        };
        let mut resolved = self
            .resolve_index_import(resolve_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_index_import(resolve_request, options).await?;
            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
        }
        if operation.current_state == OperationState::Succeeded && resolved.destination.is_none() {
            return Err(PortabilityWorkflowError::MissingDestination);
        }
        Ok(IndexImportExecutionV1 {
            operation,
            destination: resolved.destination,
        })
    }

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