lodestone-client 0.0.31

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

use lodestone_wire::{
    ApplyIndexChangeRequest, IndexChangeApplyReceipt, IndexChangeResolutionState, ResolveIndexChangeRequest,
    ResolveIndexChangeResult,
};

use crate::{CallOptions, Client, ClientError, SubmissionDisposition};

#[derive(Clone, Debug, PartialEq)]
pub struct IndexChangeExecution {
    pub receipt: IndexChangeApplyReceipt,
    pub recovered: bool,
}

#[derive(Clone, Debug, PartialEq)]
pub struct IndexChangeRecovery {
    request: ApplyIndexChangeRequest,
    resolution: Option<ResolveIndexChangeResult>,
}

impl IndexChangeRecovery {
    #[must_use]
    pub const fn request(&self) -> &ApplyIndexChangeRequest {
        &self.request
    }

    #[must_use]
    pub const fn resolution(&self) -> Option<&ResolveIndexChangeResult> {
        self.resolution.as_ref()
    }
}

#[derive(Debug)]
pub enum IndexChangeRecoveryError {
    InvalidInput,
    InvalidCoreOutcome,
    /// The original apply call authoritatively did not submit. Retrying the
    /// caller's exact request is admitted.
    Client(ClientError),
    /// The catalog authority requires the exact original action to be replayed.
    ExactReplayRequired(Box<IndexChangeRecovery>),
    /// A read-only resolver failed after submission may have occurred.
    Indeterminate(Box<IndexChangeRecovery>),
    Failed(Box<IndexChangeRecovery>),
    KnownRetired(Box<IndexChangeRecovery>),
    ExpiredUnknown(Box<IndexChangeRecovery>),
}

impl Client {
    /// Applies one validated index change and resolves any response loss by the
    /// original action identity. Success always carries the exact durable
    /// receipt; a nonterminal result preserves the request needed for replay.
    pub async fn execute_index_change(
        &self,
        request: ApplyIndexChangeRequest,
        options: CallOptions,
    ) -> Result<IndexChangeExecution, IndexChangeRecoveryError> {
        validate_request(&request)?;
        self.submit_exact_index_change(request, options, false).await
    }

    /// Replays only the exact request retained from a catalog-authoritative
    /// `not_submitted`, `prepared`, or `pending` resolution.
    pub async fn replay_index_change(
        &self,
        recovery: IndexChangeRecovery,
        options: CallOptions,
    ) -> Result<IndexChangeExecution, IndexChangeRecoveryError> {
        validate_recovery(&recovery)?;
        if !recovery.resolution.as_ref().is_some_and(|resolution| {
            matches!(
                resolution.state,
                IndexChangeResolutionState::NotSubmitted
                    | IndexChangeResolutionState::Prepared
                    | IndexChangeResolutionState::Pending
            )
        }) {
            return Err(IndexChangeRecoveryError::InvalidInput);
        }
        self.submit_exact_index_change(recovery.request, options, true).await
    }

    /// Continues read-only resolution after the prior resolver call failed.
    pub async fn resume_index_change(
        &self,
        recovery: IndexChangeRecovery,
        options: CallOptions,
    ) -> Result<IndexChangeExecution, IndexChangeRecoveryError> {
        validate_recovery(&recovery)?;
        self.resolve_exact_index_change(recovery, options, true).await
    }

    async fn submit_exact_index_change(
        &self,
        request: ApplyIndexChangeRequest,
        options: CallOptions,
        recovered: bool,
    ) -> Result<IndexChangeExecution, IndexChangeRecoveryError> {
        match self.apply_index_change(request.clone(), options.clone()).await {
            Ok(result) => {
                validate_receipt(&request, &result.receipt)?;
                Ok(IndexChangeExecution {
                    receipt: result.receipt,
                    recovered,
                })
            }
            Err(error) if !recovered && error.submission_disposition() == SubmissionDisposition::NotSubmitted => {
                Err(IndexChangeRecoveryError::Client(error))
            }
            Err(_) => {
                self.resolve_exact_index_change(
                    IndexChangeRecovery {
                        request,
                        resolution: None,
                    },
                    options,
                    true,
                )
                .await
            }
        }
    }

    async fn resolve_exact_index_change(
        &self,
        mut recovery: IndexChangeRecovery,
        options: CallOptions,
        recovered: bool,
    ) -> Result<IndexChangeExecution, IndexChangeRecoveryError> {
        let resolution = self
            .resolve_index_change(resolve_request(&recovery.request), options.for_resolution())
            .await;
        recovery.resolution = match resolution {
            Ok(resolution) => Some(resolution),
            Err(_) => return Err(IndexChangeRecoveryError::Indeterminate(Box::new(recovery))),
        };
        classify_resolution(recovery, recovered)
    }
}

fn resolve_request(request: &ApplyIndexChangeRequest) -> ResolveIndexChangeRequest {
    ResolveIndexChangeRequest {
        action_id: request.action_id.clone(),
        index: request.index.clone(),
        expected_revision: request.expected_revision.clone(),
        mapping: request.mapping.clone(),
        policies: request.policies.clone(),
    }
}

fn validate_request(request: &ApplyIndexChangeRequest) -> Result<(), IndexChangeRecoveryError> {
    if request.action_id.validate().is_err()
        || request.index.kind != "index"
        || request.index.name.is_empty()
        || request.index.incarnation.validate().is_err()
        || request.expected_revision.validate().is_err()
        || (request.mapping.is_some() == request.policies.is_some())
    {
        return Err(IndexChangeRecoveryError::InvalidInput);
    }
    if request
        .mapping
        .as_ref()
        .is_some_and(|mapping| mapping.generation.validate().is_err())
    {
        return Err(IndexChangeRecoveryError::InvalidInput);
    }
    Ok(())
}

fn validate_recovery(recovery: &IndexChangeRecovery) -> Result<(), IndexChangeRecoveryError> {
    validate_request(&recovery.request)?;
    if let Some(resolution) = &recovery.resolution {
        validate_resolution_shape(resolution)?;
        if let Some(receipt) = &resolution.receipt {
            validate_receipt(&recovery.request, receipt)?;
        }
    }
    Ok(())
}

fn validate_receipt(
    request: &ApplyIndexChangeRequest,
    receipt: &IndexChangeApplyReceipt,
) -> Result<(), IndexChangeRecoveryError> {
    let expected_revision = request
        .expected_revision
        .value
        .parse::<u64>()
        .ok()
        .and_then(|revision| revision.checked_add(1))
        .map(|revision| revision.to_string())
        .ok_or(IndexChangeRecoveryError::InvalidCoreOutcome)?;
    let expected_state = if request.mapping.is_some() {
        "mapping_applied"
    } else {
        "policy_applied"
    };
    if receipt.action_id != request.action_id
        || receipt.semantic_digest != request.action_id
        || receipt.index != request.index
        || receipt.revision.value != expected_revision
        || receipt.state != expected_state
        || receipt.schema_generation.validate().is_err()
        || request
            .mapping
            .as_ref()
            .is_some_and(|mapping| mapping.generation != receipt.schema_generation)
    {
        return Err(IndexChangeRecoveryError::InvalidCoreOutcome);
    }
    Ok(())
}

fn validate_resolution_shape(resolution: &ResolveIndexChangeResult) -> Result<(), IndexChangeRecoveryError> {
    let valid = match resolution.state {
        IndexChangeResolutionState::Applied => resolution.receipt.is_some() && resolution.failure.is_none(),
        IndexChangeResolutionState::Failed => resolution.receipt.is_none() && resolution.failure.is_some(),
        IndexChangeResolutionState::NotSubmitted
        | IndexChangeResolutionState::Prepared
        | IndexChangeResolutionState::Pending
        | IndexChangeResolutionState::KnownRetired
        | IndexChangeResolutionState::ExpiredUnknown => resolution.receipt.is_none() && resolution.failure.is_none(),
    };
    valid.then_some(()).ok_or(IndexChangeRecoveryError::InvalidCoreOutcome)
}

fn classify_resolution(
    recovery: IndexChangeRecovery,
    recovered: bool,
) -> Result<IndexChangeExecution, IndexChangeRecoveryError> {
    validate_recovery(&recovery)?;
    let resolution = recovery
        .resolution
        .as_ref()
        .ok_or(IndexChangeRecoveryError::InvalidCoreOutcome)?;
    match resolution.state {
        IndexChangeResolutionState::Applied => Ok(IndexChangeExecution {
            receipt: resolution
                .receipt
                .clone()
                .ok_or(IndexChangeRecoveryError::InvalidCoreOutcome)?,
            recovered,
        }),
        IndexChangeResolutionState::NotSubmitted
        | IndexChangeResolutionState::Prepared
        | IndexChangeResolutionState::Pending => Err(IndexChangeRecoveryError::ExactReplayRequired(Box::new(recovery))),
        IndexChangeResolutionState::Failed => Err(IndexChangeRecoveryError::Failed(Box::new(recovery))),
        IndexChangeResolutionState::KnownRetired => Err(IndexChangeRecoveryError::KnownRetired(Box::new(recovery))),
        IndexChangeResolutionState::ExpiredUnknown => Err(IndexChangeRecoveryError::ExpiredUnknown(Box::new(recovery))),
    }
}

impl Display for IndexChangeRecoveryError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::InvalidInput => "index change request is not a valid current Lodestone API action",
            Self::InvalidCoreOutcome => "Core changed the index action identity or recovery shape",
            Self::Client(_) => "Core index change was not submitted",
            Self::ExactReplayRequired(_) => "Core index change requires exact replay of the original action",
            Self::Indeterminate(_) => "Core index change outcome remains indeterminate",
            Self::Failed(_) => "Core index change reached a terminal failed receipt",
            Self::KnownRetired(_) => "Core index incarnation is authoritatively retired",
            Self::ExpiredUnknown(_) => "Core index change outcome evidence expired",
        })
    }
}

impl Error for IndexChangeRecoveryError {
    fn source(&self) -> Option<&(dyn Error + 'static)> {
        match self {
            Self::Client(error) => Some(error),
            Self::InvalidInput
            | Self::InvalidCoreOutcome
            | Self::ExactReplayRequired(_)
            | Self::Indeterminate(_)
            | Self::Failed(_)
            | Self::KnownRetired(_)
            | Self::ExpiredUnknown(_) => None,
        }
    }
}

#[cfg(test)]
mod tests {
    use lodestone_wire::{Digest, Identity, Resource, U64};

    use super::*;

    fn request() -> ApplyIndexChangeRequest {
        ApplyIndexChangeRequest {
            action_id: Digest { value: "a".repeat(64) },
            index: Resource {
                kind: "index".to_owned(),
                name: "documents".to_owned(),
                incarnation: Identity { value: "b".repeat(32) },
            },
            expected_revision: U64 { value: "7".to_owned() },
            mapping: None,
            policies: Some(lodestone_wire::IndexPolicies {
                durability: "regional_object".to_owned(),
                storage: "standard".to_owned(),
                cache: "cold_tolerant".to_owned(),
                default_visibility: Some("applied".to_owned()),
                field_promotion: lodestone_wire::FieldPromotionConfiguration {
                    format_version: 3,
                    enabled: false,
                    minimum_query_count: 1,
                    maximum_promoted_fields: 1,
                    maximum_cardinality: U64 { value: "1".to_owned() },
                    maximum_extra_bytes: U64 { value: "1".to_owned() },
                    allow_sensitive: false,
                },
                overload: lodestone_wire::OverloadConfiguration {
                    format_version: 3,
                    mode: "reject".to_owned(),
                    maximum_backlog_bytes: U64 { value: "1".to_owned() },
                    maximum_backlog_age_millis: U64 { value: "1".to_owned() },
                    maximum_collection_share_bytes: U64 { value: "1".to_owned() },
                    maximum_recovery_ops_per_pass: 1,
                    maximum_pending_overlay_documents: U64 { value: "1".to_owned() },
                },
            }),
        }
    }

    #[test]
    fn pending_resolution_requires_exact_identity_replay() -> Result<(), Box<dyn Error>> {
        let recovery = IndexChangeRecovery {
            request: request(),
            resolution: Some(ResolveIndexChangeResult {
                state: IndexChangeResolutionState::Pending,
                receipt: None,
                failure: None,
            }),
        };
        match classify_resolution(recovery, true) {
            Err(IndexChangeRecoveryError::ExactReplayRequired(recovery)) => {
                assert_eq!(recovery.request().action_id.value, "a".repeat(64));
                Ok(())
            }
            _ => Err("pending resolution did not require exact replay".into()),
        }
    }

    #[test]
    fn applied_resolution_requires_exact_successor_receipt() -> Result<(), Box<dyn Error>> {
        let request = request();
        let receipt = IndexChangeApplyReceipt {
            action_id: request.action_id.clone(),
            semantic_digest: request.action_id.clone(),
            index: request.index.clone(),
            revision: U64 { value: "8".to_owned() },
            schema_generation: U64 { value: "3".to_owned() },
            state: "policy_applied".to_owned(),
        };
        let execution = classify_resolution(
            IndexChangeRecovery {
                request,
                resolution: Some(ResolveIndexChangeResult {
                    state: IndexChangeResolutionState::Applied,
                    receipt: Some(receipt.clone()),
                    failure: None,
                }),
            },
            true,
        )?;
        assert_eq!(execution.receipt, receipt);
        assert!(execution.recovered);
        Ok(())
    }
}