lodestone-client 0.0.25

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

use lodestone_wire::{
    BulkDocumentItem, BulkDocumentItemOutcome, BulkDocumentOutcome, Digest, PrepareDocumentBulkRequest,
    ResolveDocumentBulkRequest, Resource, SubmissionState, SubmitDocumentBulkRequest,
};

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

#[derive(Clone, Debug, PartialEq)]
pub struct BulkExecution {
    pub outcome: BulkDocumentOutcome,
    pub recovered: bool,
}

#[derive(Clone, PartialEq)]
pub struct BulkRecovery {
    batch_id: Digest,
    index: Resource,
    items: Vec<BulkDocumentItem>,
    outcome: BulkDocumentOutcome,
}

impl BulkRecovery {
    #[must_use]
    pub const fn batch_id(&self) -> &Digest {
        &self.batch_id
    }

    #[must_use]
    pub const fn index(&self) -> &Resource {
        &self.index
    }

    #[must_use]
    pub fn items(&self) -> &[BulkDocumentItem] {
        &self.items
    }

    #[must_use]
    pub const fn outcome(&self) -> &BulkDocumentOutcome {
        &self.outcome
    }
}

impl fmt::Debug for BulkRecovery {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("BulkRecovery")
            .field("batch_id", &self.batch_id)
            .field("index", &self.index)
            .field("item_count", &self.items.len())
            .field("outcome", &self.outcome)
            .finish()
    }
}

#[derive(Debug)]
pub enum BulkRecoveryError {
    InvalidInput,
    InvalidCoreOutcome,
    /// Preparation failed, or submission authoritatively reported that it did
    /// not start. Replaying the caller's original batch is admitted.
    Client(ClientError),
    /// Submission may have started and the last authoritative bulk outcome
    /// did not reach completion. Resolver-call failures stay in this phase
    /// even when the read-only resolver itself reports `NotSubmitted`.
    Indeterminate(Box<BulkRecovery>),
    /// Resolution proved that at least one exact item was not submitted while
    /// every other item is resolved or also not submitted. Replay must use
    /// this recovery value so the batch, action, and locator identities stay
    /// unchanged.
    ExactReplayRequired(Box<BulkRecovery>),
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ResolutionProgress {
    Resolved,
    ExactReplayRequired,
    Pending,
}

impl Client {
    /// Prepares every item, submits the batch at most once, and resolves any
    /// ambiguous or partial outcome by the stable per-item locators. Success
    /// means every item is authoritatively resolved.
    pub async fn execute_document_bulk(
        &self,
        batch_id: Digest,
        index: Resource,
        items: Vec<BulkDocumentItem>,
        options: CallOptions,
    ) -> Result<BulkExecution, BulkRecoveryError> {
        validate_bulk_input(&batch_id, &index, &items)?;
        let prepared = self
            .prepare_document_bulk(
                PrepareDocumentBulkRequest {
                    batch_id: batch_id.clone(),
                    index: index.clone(),
                    items: items.clone(),
                },
                options.clone(),
            )
            .await
            .map_err(BulkRecoveryError::Client)?
            .outcome;
        validate_bulk_identity(&prepared, &batch_id, &index, &items)?;
        if !prepared.complete
            || prepared.items.iter().any(|item| {
                item.state != SubmissionState::NotSubmitted
                    || item.locator.is_none()
                    || item.sequence.is_some()
                    || item.mapping_generation.is_some()
                    || item.mapping_adopted
                    || item.no_document_effect
                    || item.error.is_some()
            })
        {
            return Err(BulkRecoveryError::InvalidCoreOutcome);
        }

        let preparations = prepared.items.clone();
        let recovery = BulkRecovery {
            batch_id,
            index,
            items,
            outcome: prepared,
        };
        self.submit_exact_document_bulk(recovery, preparations, options, false)
            .await
    }

    /// Retries only the exact batch, item actions, semantics, and locators
    /// returned by an authoritative terminal resolution containing one or more
    /// `NotSubmitted` items.
    pub async fn replay_document_bulk(
        &self,
        recovery: BulkRecovery,
        options: CallOptions,
    ) -> Result<BulkExecution, BulkRecoveryError> {
        validate_bulk_recovery(&recovery)?;
        if resolution_progress(&recovery.outcome)? != ResolutionProgress::ExactReplayRequired {
            return Err(BulkRecoveryError::InvalidInput);
        }
        let preparations = exact_replay_preparations(&recovery.outcome)?;
        self.submit_exact_document_bulk(recovery, preparations, options, true)
            .await
    }

    /// Continues read-only resolution for an exact possibly submitted batch.
    pub async fn resume_document_bulk(
        &self,
        recovery: BulkRecovery,
        options: CallOptions,
    ) -> Result<BulkExecution, BulkRecoveryError> {
        validate_bulk_recovery(&recovery)?;
        self.resolve_exact_document_bulk(recovery, options, true).await
    }

    async fn submit_exact_document_bulk(
        &self,
        mut recovery: BulkRecovery,
        preparations: Vec<BulkDocumentItemOutcome>,
        options: CallOptions,
        exact_replay: bool,
    ) -> Result<BulkExecution, BulkRecoveryError> {
        let submitted = self
            .submit_document_bulk(
                SubmitDocumentBulkRequest {
                    batch_id: recovery.batch_id.clone(),
                    index: recovery.index.clone(),
                    items: recovery.items.clone(),
                    preparations,
                },
                options.clone(),
            )
            .await;
        match submitted {
            Ok(result) => {
                validate_bulk_identity(&result.outcome, &recovery.batch_id, &recovery.index, &recovery.items)?;
                validate_submit_completion(&result.outcome)?;
                if result.outcome.complete {
                    return Ok(BulkExecution {
                        outcome: result.outcome,
                        recovered: exact_replay,
                    });
                }
                recovery.outcome = result.outcome;
            }
            Err(error) if error.submission_disposition() != SubmissionDisposition::NotSubmitted => {}
            Err(_error) if exact_replay => {
                return Err(BulkRecoveryError::ExactReplayRequired(Box::new(recovery)));
            }
            Err(error) => return Err(BulkRecoveryError::Client(error)),
        }

        self.resolve_exact_document_bulk(recovery, options, true).await
    }

    async fn resolve_exact_document_bulk(
        &self,
        mut recovery: BulkRecovery,
        options: CallOptions,
        recovered: bool,
    ) -> Result<BulkExecution, BulkRecoveryError> {
        let maximum_resolutions = usize::from(self.pool_config().maximum_read_retries).saturating_add(1);
        for _ in 0..maximum_resolutions {
            let previous = recovery.outcome.clone();
            let resolution = self
                .resolve_document_bulk(
                    ResolveDocumentBulkRequest {
                        batch_id: recovery.batch_id.clone(),
                        index: recovery.index.clone(),
                        // Keep the last authoritative observation intact so a
                        // failed resolver can return exact recovery state.
                        items: recovery.outcome.items.clone(),
                    },
                    options.for_resolution(),
                )
                .await;
            recovery.outcome = match resolution {
                Ok(result) => result.outcome,
                Err(error) => return Err(indeterminate_resolution_failure(recovery, error)),
            };
            validate_bulk_recovery(&recovery)?;
            match resolution_progress(&recovery.outcome)? {
                ResolutionProgress::Resolved => {
                    return Ok(BulkExecution {
                        outcome: recovery.outcome,
                        recovered,
                    });
                }
                ResolutionProgress::ExactReplayRequired => {
                    return Err(BulkRecoveryError::ExactReplayRequired(Box::new(recovery)));
                }
                ResolutionProgress::Pending => {}
            }
            if recovery.outcome == previous {
                break;
            }
        }
        Err(BulkRecoveryError::Indeterminate(Box::new(recovery)))
    }
}

fn indeterminate_resolution_failure(recovery: BulkRecovery, _resolver_error: ClientError) -> BulkRecoveryError {
    // INVARIANT: the resolver describes an already possibly submitted batch.
    // Its read-only call disposition cannot downgrade the original mutation to
    // a caller-retryable pre-submission failure.
    BulkRecoveryError::Indeterminate(Box::new(recovery))
}

fn validate_bulk_recovery(recovery: &BulkRecovery) -> Result<(), BulkRecoveryError> {
    validate_bulk_input(&recovery.batch_id, &recovery.index, &recovery.items)?;
    validate_bulk_identity(&recovery.outcome, &recovery.batch_id, &recovery.index, &recovery.items)
}

fn exact_replay_preparations(outcome: &BulkDocumentOutcome) -> Result<Vec<BulkDocumentItemOutcome>, BulkRecoveryError> {
    outcome
        .items
        .iter()
        .map(|item| {
            if !matches!(item.state, SubmissionState::NotSubmitted | SubmissionState::Resolved) {
                return Err(BulkRecoveryError::InvalidCoreOutcome);
            }
            let locator = item.locator.clone().ok_or(BulkRecoveryError::InvalidCoreOutcome)?;
            Ok(BulkDocumentItemOutcome {
                item_action_id: item.item_action_id.clone(),
                locator: Some(locator),
                state: SubmissionState::NotSubmitted,
                sequence: None,
                mapping_generation: None,
                mapping_adopted: false,
                no_document_effect: false,
                error: None,
            })
        })
        .collect()
}

fn validate_bulk_input(
    batch_id: &Digest,
    index: &Resource,
    items: &[BulkDocumentItem],
) -> Result<(), BulkRecoveryError> {
    if batch_id.validate().is_err()
        || index.kind != "index"
        || index.name.is_empty()
        || index.incarnation.validate().is_err()
        || items.is_empty()
        || items.len() > 1_000
        || items.iter().any(|item| item.item_action_id.validate().is_err())
    {
        return Err(BulkRecoveryError::InvalidInput);
    }
    let mut action_ids = std::collections::BTreeSet::new();
    if items
        .iter()
        .any(|item| !action_ids.insert(item.item_action_id.value.as_str()))
    {
        return Err(BulkRecoveryError::InvalidInput);
    }
    Ok(())
}

fn validate_bulk_identity(
    outcome: &BulkDocumentOutcome,
    batch_id: &Digest,
    index: &Resource,
    items: &[BulkDocumentItem],
) -> Result<(), BulkRecoveryError> {
    if &outcome.batch_id != batch_id || &outcome.index != index || outcome.items.len() != items.len() {
        return Err(BulkRecoveryError::InvalidCoreOutcome);
    }
    for (expected, observed) in items.iter().zip(&outcome.items) {
        if observed.item_action_id != expected.item_action_id || !valid_item_outcome(observed) {
            return Err(BulkRecoveryError::InvalidCoreOutcome);
        }
    }
    Ok(())
}

fn valid_item_outcome(item: &BulkDocumentItemOutcome) -> bool {
    item.item_action_id.validate().is_ok()
        && match item.state {
            SubmissionState::NotSubmitted => item.sequence.is_none(),
            SubmissionState::Submitted | SubmissionState::Indeterminate => item.locator.is_some(),
            SubmissionState::Resolved => item.locator.is_some(),
            SubmissionState::KnownRetired => item.locator.is_some() && item.error.is_some(),
        }
}

fn validate_submit_completion(outcome: &BulkDocumentOutcome) -> Result<(), BulkRecoveryError> {
    let fully_resolved = outcome.items.iter().all(|item| item.state == SubmissionState::Resolved);
    if outcome.complete != fully_resolved {
        return Err(BulkRecoveryError::InvalidCoreOutcome);
    }
    Ok(())
}

fn validate_resolution_completion(outcome: &BulkDocumentOutcome) -> Result<(), BulkRecoveryError> {
    let terminal = outcome
        .items
        .iter()
        .all(|item| matches!(item.state, SubmissionState::NotSubmitted | SubmissionState::Resolved));
    if outcome.complete != terminal {
        return Err(BulkRecoveryError::InvalidCoreOutcome);
    }
    Ok(())
}

fn resolution_progress(outcome: &BulkDocumentOutcome) -> Result<ResolutionProgress, BulkRecoveryError> {
    validate_resolution_completion(outcome)?;
    if outcome.items.iter().all(|item| item.state == SubmissionState::Resolved) {
        return Ok(ResolutionProgress::Resolved);
    }
    if outcome.complete {
        return Ok(ResolutionProgress::ExactReplayRequired);
    }
    Ok(ResolutionProgress::Pending)
}

impl Display for BulkRecoveryError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::InvalidInput => "bulk request is not a valid current Lodestone API batch",
            Self::InvalidCoreOutcome => "Core changed the bulk identity, order, or submission shape",
            Self::Client(_) => "Core bulk execution failed",
            Self::Indeterminate(_) => "Core bulk outcome remains indeterminate after submission may have occurred",
            Self::ExactReplayRequired(_) => {
                "Core bulk outcome requires exact replay of the original batch and item identities"
            }
        })
    }
}

impl Error for BulkRecoveryError {}

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

    use super::*;

    fn outcome(states: &[SubmissionState], complete: bool) -> BulkDocumentOutcome {
        BulkDocumentOutcome {
            batch_id: Digest { value: "a".repeat(64) },
            index: Resource {
                kind: "index".to_owned(),
                name: "events".to_owned(),
                incarnation: Identity { value: "b".repeat(32) },
            },
            items: states
                .iter()
                .enumerate()
                .map(|(ordinal, state)| BulkDocumentItemOutcome {
                    item_action_id: Digest {
                        value: format!("{:064x}", ordinal + 1),
                    },
                    locator: Some(Locator {
                        authority: "document".to_owned(),
                        value: format!("locator-{ordinal}"),
                    }),
                    state: *state,
                    sequence: None,
                    mapping_generation: None,
                    mapping_adopted: false,
                    no_document_effect: false,
                    error: None,
                })
                .collect(),
            complete,
            achievement: None,
        }
    }

    fn recovery(states: &[SubmissionState], complete: bool) -> BulkRecovery {
        let outcome = outcome(states, complete);
        BulkRecovery {
            batch_id: outcome.batch_id.clone(),
            index: outcome.index.clone(),
            items: outcome
                .items
                .iter()
                .enumerate()
                .map(|(ordinal, item)| BulkDocumentItem {
                    item_action_id: item.item_action_id.clone(),
                    document_id: format!("document-{ordinal}"),
                    operation: "delete".to_owned(),
                    source: None,
                    record: None,
                    routing: None,
                    condition: None,
                    pipeline: None,
                    semantic_digest: Digest {
                        value: format!("{:064x}", ordinal + 100),
                    },
                })
                .collect(),
            outcome,
        }
    }

    #[test]
    fn partial_submit_is_recovered_instead_of_rejected_as_a_false_completion() {
        let partial = outcome(&[SubmissionState::Resolved, SubmissionState::NotSubmitted], false);
        assert!(validate_submit_completion(&partial).is_ok());
        assert!(validate_resolution_completion(&partial).is_err());

        let false_complete = outcome(&[SubmissionState::NotSubmitted], true);
        assert!(validate_submit_completion(&false_complete).is_err());
    }

    #[test]
    fn resolver_failure_preserves_the_last_authoritative_outcome_as_indeterminate() {
        let authoritative = recovery(&[SubmissionState::Indeterminate], false);
        let resolver_error = ClientError::ProfileNotVerified;
        assert_eq!(
            resolver_error.submission_disposition(),
            SubmissionDisposition::NotSubmitted
        );
        let error = indeterminate_resolution_failure(authoritative.clone(), resolver_error);

        assert!(matches!(&error, BulkRecoveryError::Indeterminate(_)));
        if let BulkRecoveryError::Indeterminate(observed) = error {
            assert_eq!(*observed, authoritative);
        }
    }

    #[test]
    fn all_not_submitted_resolution_requires_exact_replay_instead_of_success() -> Result<(), BulkRecoveryError> {
        let outcome = outcome(&[SubmissionState::NotSubmitted, SubmissionState::NotSubmitted], true);
        assert_eq!(resolution_progress(&outcome)?, ResolutionProgress::ExactReplayRequired);
        Ok(())
    }

    #[test]
    fn mixed_resolved_and_not_submitted_resolution_requires_exact_replay() -> Result<(), BulkRecoveryError> {
        let outcome = outcome(&[SubmissionState::Resolved, SubmissionState::NotSubmitted], true);
        assert_eq!(resolution_progress(&outcome)?, ResolutionProgress::ExactReplayRequired);
        let preparations = exact_replay_preparations(&outcome)?;
        assert_eq!(preparations.len(), outcome.items.len());
        for (replayed, authoritative) in preparations.iter().zip(&outcome.items) {
            assert_eq!(replayed.item_action_id, authoritative.item_action_id);
            assert_eq!(replayed.locator, authoritative.locator);
            assert_eq!(replayed.state, SubmissionState::NotSubmitted);
            assert!(replayed.sequence.is_none());
        }
        Ok(())
    }
}