lodestone-client 0.0.4

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(Debug)]
pub enum BulkRecoveryError {
    InvalidInput,
    InvalidCoreOutcome,
    Client(ClientError),
    Indeterminate(Box<BulkDocumentOutcome>),
}

impl Client {
    /// Prepares every item, submits the batch at most once, and resolves any
    /// ambiguous or partial outcome by the stable per-item locators.
    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.error.is_some()
            })
        {
            return Err(BulkRecoveryError::InvalidCoreOutcome);
        }

        let submitted = self
            .submit_document_bulk(
                SubmitDocumentBulkRequest {
                    batch_id: batch_id.clone(),
                    index: index.clone(),
                    items: items.clone(),
                    preparations: prepared.items.clone(),
                },
                options.clone(),
            )
            .await;
        let (mut outcome, recovered) = match submitted {
            Ok(result) => {
                validate_bulk_identity(&result.outcome, &batch_id, &index, &items)?;
                validate_submit_completion(&result.outcome)?;
                if result.outcome.complete {
                    return Ok(BulkExecution {
                        outcome: result.outcome,
                        recovered: false,
                    });
                }
                (result.outcome, true)
            }
            Err(error) if error.submission_disposition() != SubmissionDisposition::NotSubmitted => (prepared, true),
            Err(error) => return Err(BulkRecoveryError::Client(error)),
        };

        let maximum_resolutions = usize::from(self.pool_config().maximum_read_retries).saturating_add(1);
        for _ in 0..maximum_resolutions {
            let previous = outcome.clone();
            let resolution = self
                .resolve_document_bulk(
                    ResolveDocumentBulkRequest {
                        batch_id: batch_id.clone(),
                        index: index.clone(),
                        // Keep the last authoritative observation intact so a
                        // deadline can return its full indeterminate outcome.
                        items: outcome.items.clone(),
                    },
                    options.for_resolution(),
                )
                .await;
            outcome = match resolution {
                Ok(result) => result.outcome,
                Err(error) if error.is_deadline_exceeded() => {
                    return Err(BulkRecoveryError::Indeterminate(Box::new(outcome)));
                }
                Err(error) => return Err(BulkRecoveryError::Client(error)),
            };
            validate_bulk_identity(&outcome, &batch_id, &index, &items)?;
            validate_resolution_completion(&outcome)?;
            if outcome.complete {
                return Ok(BulkExecution { outcome, recovered });
            }
            if outcome == previous {
                break;
            }
        }
        Err(BulkRecoveryError::Indeterminate(Box::new(outcome)))
    }
}

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

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 bounded resolution",
        })
    }
}

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!("{ordinal:064x}"),
                    },
                    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,
        }
    }

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