lodestone-client 0.0.4

//! Closed, secret-free recovery model for retained append provisioning.

use aws_lc_rs::digest;
use lodestone_wire::{
    Digest, Identity, IndexResourceSpec, LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION, Locator, MappingMode,
    PrepareCreateIndexRequest, Resource,
};
use serde::{Deserialize, Serialize};

use crate::{
    LocalOperatorWorkflowKind, ValidatedConnectionProfile, WorkflowJournalError, WorkflowJournalIntent,
    WorkflowJournalTransition,
};

pub const PROVISION_RETAINED_APPEND_FORMAT_VERSION: u16 = 1;
const MAX_NAME_BYTES: usize = 255;
const MAX_ROLE_NAME_BYTES: usize = 128;

/// Immutable, secret-free inputs identifying one provisioning workflow.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProvisionRetainedAppendIntent {
    pub format_version: u16,
    pub cluster_id: Identity,
    pub collection_id: Identity,
    pub admin_profile_sha256: [u8; 32],
    pub consumer_profile_sha256: [u8; 32],
    pub trust_sha256: [u8; 32],
    pub definition_sha256: [u8; 32],
    pub index_name: String,
    pub role_name: String,
}

impl ProvisionRetainedAppendIntent {
    pub fn new(
        cluster_id: Identity,
        collection_id: Identity,
        admin_profile: &ValidatedConnectionProfile,
        consumer_profile: &ValidatedConnectionProfile,
        trust_sha256: [u8; 32],
        definition: &PrepareCreateIndexRequest,
        role_name: String,
    ) -> Result<Self, WorkflowJournalError> {
        let intent = Self {
            format_version: PROVISION_RETAINED_APPEND_FORMAT_VERSION,
            cluster_id,
            collection_id,
            admin_profile_sha256: profile_sha256(admin_profile)?,
            consumer_profile_sha256: profile_sha256(consumer_profile)?,
            trust_sha256,
            definition_sha256: provision_definition_sha256(definition)?,
            index_name: definition.name.clone(),
            role_name,
        };
        intent.validate()?;
        Ok(intent)
    }
}

impl crate::workflow_journal::sealed::Sealed for ProvisionRetainedAppendIntent {}

impl WorkflowJournalIntent for ProvisionRetainedAppendIntent {
    type Transition = ProvisionRetainedAppendTransition;

    const KIND: LocalOperatorWorkflowKind = LocalOperatorWorkflowKind::ProvisionRetainedAppend;

    fn validate(&self) -> Result<(), WorkflowJournalError> {
        if self.format_version != PROVISION_RETAINED_APPEND_FORMAT_VERSION
            || !valid_identity(&self.cluster_id)
            || !valid_identity(&self.collection_id)
            || !valid_sha256(&self.admin_profile_sha256)
            || !valid_sha256(&self.consumer_profile_sha256)
            || !valid_sha256(&self.trust_sha256)
            || !valid_sha256(&self.definition_sha256)
            || !valid_index_name(&self.index_name)
            || !valid_name(&self.role_name, MAX_ROLE_NAME_BYTES)
        {
            return Err(WorkflowJournalError::InvalidIntent);
        }
        Ok(())
    }
}

/// How a prepared role action became known to have applied.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum RoleAppendApplyResolution {
    SubmitResponse,
    Resolver,
}

/// Why one prepared locator is safe to stop submitting before a new prepare.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum RoleAppendAttemptClosure {
    ExpiredUnknown,
    SupersededNoEffect,
}

/// Monotonic, secret-free observations committed by the workflow.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
pub enum ProvisionRetainedAppendTransition {
    PreflightVerified,
    CatalogPrepared {
        locator: Locator,
        semantic_digest: Digest,
        expires_at: lodestone_wire::U64,
        resource_spec: IndexResourceSpec,
    },
    CatalogApplied {
        locator: Locator,
        resource: Resource,
    },
    IndexActive {
        resource: Resource,
        index_revision: lodestone_wire::U64,
    },
    RolePrepared {
        locator: Locator,
        semantic_digest: Digest,
        expires_at: lodestone_wire::U64,
        predecessor_role_version: lodestone_wire::U64,
        successor_role_version: lodestone_wire::U64,
    },
    RoleAttemptClosed {
        locator: Locator,
        closure: RoleAppendAttemptClosure,
    },
    RoleApplied {
        locator: Locator,
        resolution: RoleAppendApplyResolution,
        target_policy_revision: lodestone_wire::U64,
    },
    RoleVerified {
        resource: Resource,
        role_version: lodestone_wire::U64,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        target_policy_revision: Option<lodestone_wire::U64>,
    },
    DiscoveryVerified {
        resource: Resource,
        observed_policy_revision: lodestone_wire::U64,
    },
    Completed {
        resource: Resource,
        role_version: lodestone_wire::U64,
        observed_policy_revision: lodestone_wire::U64,
    },
}

#[derive(Clone, Debug, PartialEq)]
pub struct ProvisionCatalogPreparation {
    pub locator: Locator,
    pub semantic_digest: Digest,
    pub expires_at: lodestone_wire::U64,
    pub resource_spec: IndexResourceSpec,
}

#[derive(Clone, Debug, PartialEq)]
pub struct ProvisionRolePreparation {
    pub locator: Locator,
    pub semantic_digest: Digest,
    pub expires_at: lodestone_wire::U64,
    pub predecessor_role_version: lodestone_wire::U64,
    pub successor_role_version: lodestone_wire::U64,
}

impl crate::workflow_journal::sealed::Sealed for ProvisionRetainedAppendTransition {}

impl WorkflowJournalTransition for ProvisionRetainedAppendTransition {
    fn validate(&self) -> Result<(), WorkflowJournalError> {
        let valid = match self {
            Self::PreflightVerified => true,
            Self::CatalogPrepared {
                locator,
                semantic_digest,
                expires_at,
                resource_spec,
            } => {
                valid_locator(locator)
                    && valid_digest(semantic_digest)
                    && valid_u64(expires_at)
                    && resource_spec.format_version == u32::from(LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION)
            }
            Self::CatalogApplied { locator, resource } => valid_locator(locator) && valid_resource(resource),
            Self::IndexActive {
                resource,
                index_revision,
            } => valid_resource(resource) && valid_u64(index_revision),
            Self::RolePrepared {
                locator,
                semantic_digest,
                expires_at,
                predecessor_role_version,
                successor_role_version,
            } => {
                valid_locator(locator)
                    && valid_digest(semantic_digest)
                    && valid_u64(expires_at)
                    && ordered_successor(predecessor_role_version, successor_role_version)
            }
            Self::RoleAttemptClosed { locator, .. } => valid_locator(locator),
            Self::RoleApplied {
                locator,
                target_policy_revision,
                ..
            } => valid_locator(locator) && valid_u64(target_policy_revision),
            Self::RoleVerified {
                resource,
                role_version,
                target_policy_revision,
            } => {
                valid_resource(resource)
                    && valid_u64(role_version)
                    && target_policy_revision.as_ref().is_none_or(valid_u64)
            }
            Self::DiscoveryVerified {
                resource,
                observed_policy_revision,
            } => valid_resource(resource) && valid_u64(observed_policy_revision),
            Self::Completed {
                resource,
                role_version,
                observed_policy_revision,
            } => valid_resource(resource) && valid_u64(role_version) && valid_u64(observed_policy_revision),
        };
        if valid {
            Ok(())
        } else {
            Err(WorkflowJournalError::InvalidTransition)
        }
    }
}

/// Replayed state after validating the complete transition ordering.
#[derive(Clone, Debug, Default, PartialEq)]
pub struct ProvisionRetainedAppendState {
    preflight_verified: bool,
    catalog_preparation: Option<ProvisionCatalogPreparation>,
    catalog_resource: Option<Resource>,
    index_active: bool,
    role_preparation: Option<ProvisionRolePreparation>,
    target_policy_revision: Option<lodestone_wire::U64>,
    role_version: Option<lodestone_wire::U64>,
    discovery_policy_revision: Option<lodestone_wire::U64>,
    completed: bool,
}

impl ProvisionRetainedAppendState {
    pub fn replay(transitions: &[ProvisionRetainedAppendTransition]) -> Result<Self, WorkflowJournalError> {
        let mut state = Self::default();
        for transition in transitions {
            transition.validate()?;
            state.apply(transition)?;
        }
        Ok(state)
    }

    fn apply(&mut self, transition: &ProvisionRetainedAppendTransition) -> Result<(), WorkflowJournalError> {
        if self.completed {
            return Err(WorkflowJournalError::Corrupt);
        }
        match transition {
            ProvisionRetainedAppendTransition::PreflightVerified
                if !self.preflight_verified && self.catalog_preparation.is_none() =>
            {
                self.preflight_verified = true;
            }
            ProvisionRetainedAppendTransition::CatalogPrepared {
                locator,
                semantic_digest,
                expires_at,
                resource_spec,
            } if self.preflight_verified && self.catalog_preparation.is_none() && self.catalog_resource.is_none() => {
                self.catalog_preparation = Some(ProvisionCatalogPreparation {
                    locator: locator.clone(),
                    semantic_digest: semantic_digest.clone(),
                    expires_at: expires_at.clone(),
                    resource_spec: resource_spec.clone(),
                });
            }
            ProvisionRetainedAppendTransition::CatalogApplied { locator, resource }
                if self.catalog_resource.is_none()
                    && self
                        .catalog_preparation
                        .as_ref()
                        .is_some_and(|prepared| &prepared.locator == locator) =>
            {
                self.catalog_resource = Some(resource.clone());
            }
            ProvisionRetainedAppendTransition::IndexActive { resource, .. }
                if !self.index_active && self.catalog_resource.as_ref() == Some(resource) =>
            {
                self.index_active = true;
            }
            ProvisionRetainedAppendTransition::RolePrepared {
                locator,
                semantic_digest,
                expires_at,
                predecessor_role_version,
                successor_role_version,
            } if self.index_active && self.role_preparation.is_none() && self.role_version.is_none() => {
                self.role_preparation = Some(ProvisionRolePreparation {
                    locator: locator.clone(),
                    semantic_digest: semantic_digest.clone(),
                    expires_at: expires_at.clone(),
                    predecessor_role_version: predecessor_role_version.clone(),
                    successor_role_version: successor_role_version.clone(),
                });
            }
            ProvisionRetainedAppendTransition::RoleAttemptClosed { locator, .. }
                if self
                    .role_preparation
                    .as_ref()
                    .is_some_and(|prepared| &prepared.locator == locator)
                    && self.target_policy_revision.is_none() =>
            {
                self.role_preparation = None;
            }
            ProvisionRetainedAppendTransition::RoleApplied {
                locator,
                target_policy_revision,
                ..
            } if self
                .role_preparation
                .as_ref()
                .is_some_and(|prepared| &prepared.locator == locator)
                && self.target_policy_revision.is_none() =>
            {
                self.target_policy_revision = Some(target_policy_revision.clone());
            }
            ProvisionRetainedAppendTransition::RoleVerified {
                resource,
                role_version,
                target_policy_revision,
            } if self.index_active
                && self.catalog_resource.as_ref() == Some(resource)
                && self.role_version.is_none()
                && target_policy_revision == &self.target_policy_revision
                && self.role_version_is_admissible(role_version) =>
            {
                self.role_version = Some(role_version.clone());
            }
            ProvisionRetainedAppendTransition::DiscoveryVerified {
                resource,
                observed_policy_revision,
            } if self.catalog_resource.as_ref() == Some(resource)
                && self.role_version.is_some()
                && self.discovery_policy_revision.is_none()
                && revision_at_least(observed_policy_revision, self.target_policy_revision.as_ref()) =>
            {
                self.discovery_policy_revision = Some(observed_policy_revision.clone());
            }
            ProvisionRetainedAppendTransition::Completed {
                resource,
                role_version,
                observed_policy_revision,
            } if self.catalog_resource.as_ref() == Some(resource)
                && revision_at_least(role_version, self.role_version.as_ref())
                && self.discovery_policy_revision.as_ref() == Some(observed_policy_revision) =>
            {
                self.role_version = Some(role_version.clone());
                self.completed = true;
            }
            _ => return Err(WorkflowJournalError::Corrupt),
        }
        Ok(())
    }

    fn role_version_is_admissible(&self, observed: &lodestone_wire::U64) -> bool {
        match &self.role_preparation {
            Some(preparation) if self.target_policy_revision.is_some() => {
                revision_at_least(observed, Some(&preparation.successor_role_version))
            }
            Some(_) => false,
            None => self.target_policy_revision.is_none(),
        }
    }

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

    #[must_use]
    pub fn catalog_preparation(&self) -> Option<&ProvisionCatalogPreparation> {
        self.catalog_preparation.as_ref()
    }

    #[must_use]
    pub fn resource(&self) -> Option<&Resource> {
        self.catalog_resource.as_ref()
    }

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

    #[must_use]
    pub fn role_preparation(&self) -> Option<&ProvisionRolePreparation> {
        self.role_preparation.as_ref()
    }

    #[must_use]
    pub fn target_policy_revision(&self) -> Option<&lodestone_wire::U64> {
        self.target_policy_revision.as_ref()
    }

    #[must_use]
    pub fn role_version(&self) -> Option<&lodestone_wire::U64> {
        self.role_version.as_ref()
    }

    #[must_use]
    pub fn discovery_policy_revision(&self) -> Option<&lodestone_wire::U64> {
        self.discovery_policy_revision.as_ref()
    }

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

pub fn profile_sha256(profile: &ValidatedConnectionProfile) -> Result<[u8; 32], WorkflowJournalError> {
    let bytes = profile
        .encode_current()
        .map_err(|_| WorkflowJournalError::InvalidIntent)?;
    Ok(sha256(&bytes))
}

pub fn provision_definition_sha256(definition: &PrepareCreateIndexRequest) -> Result<[u8; 32], WorkflowJournalError> {
    if definition.mapping.mode != MappingMode::AppendLog || definition.retained_event_identity != Some(true) {
        return Err(WorkflowJournalError::InvalidIntent);
    }
    let bytes = serde_json::to_vec(definition).map_err(|_| WorkflowJournalError::Codec)?;
    Ok(sha256(&bytes))
}

fn sha256(bytes: &[u8]) -> [u8; 32] {
    let observed = digest::digest(&digest::SHA256, bytes);
    let mut output = [0; 32];
    output.copy_from_slice(observed.as_ref());
    output
}

fn valid_identity(value: &Identity) -> bool {
    value.value.len() == 32
        && value
            .value
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
        && u128::from_str_radix(&value.value, 16).is_ok_and(|value| value != 0)
}

fn valid_sha256(value: &[u8; 32]) -> bool {
    value.iter().any(|byte| *byte != 0)
}

fn valid_name(value: &str, maximum: usize) -> bool {
    !value.is_empty()
        && value.len() <= maximum
        && value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
}

fn valid_index_name(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= MAX_NAME_BYTES
        && !value
            .bytes()
            .any(|byte| byte.is_ascii_control() || byte.is_ascii_whitespace())
}

fn valid_locator(value: &Locator) -> bool {
    !value.authority.is_empty()
        && value.authority.len() <= 128
        && !value.value.is_empty()
        && value.value.len() <= 512
        && !value.authority.chars().any(char::is_control)
        && !value.value.chars().any(char::is_control)
}

fn valid_digest(value: &Digest) -> bool {
    value.value.len() == 64
        && value
            .value
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
        && value.value.bytes().any(|byte| byte != b'0')
}

fn valid_resource(value: &Resource) -> bool {
    value.kind == "index" && valid_index_name(&value.name) && valid_identity(&value.incarnation)
}

fn valid_u64(value: &lodestone_wire::U64) -> bool {
    value
        .value
        .parse::<u64>()
        .is_ok_and(|parsed| parsed.to_string() == value.value)
}

fn ordered_successor(predecessor: &lodestone_wire::U64, successor: &lodestone_wire::U64) -> bool {
    match (predecessor.value.parse::<u64>(), successor.value.parse::<u64>()) {
        (Ok(predecessor), Ok(successor)) => predecessor.checked_add(1) == Some(successor),
        _ => false,
    }
}

fn revision_at_least(observed: &lodestone_wire::U64, minimum: Option<&lodestone_wire::U64>) -> bool {
    match (
        observed.value.parse::<u64>(),
        minimum.map(|value| value.value.parse::<u64>()),
    ) {
        (Ok(_), None) => valid_u64(observed),
        (Ok(observed), Some(Ok(minimum))) => observed >= minimum,
        _ => false,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use lodestone_wire::{ResourceDurabilityClass, ResourceStorageClass, U64};

    fn id(value: char) -> Identity {
        Identity {
            value: value.to_string().repeat(32),
        }
    }

    fn digest(value: char) -> Digest {
        Digest {
            value: value.to_string().repeat(64),
        }
    }

    fn locator(value: &str) -> Locator {
        Locator {
            authority: "catalog".to_owned(),
            value: value.to_owned(),
        }
    }

    fn resource() -> Resource {
        Resource {
            kind: "index".to_owned(),
            name: "events".to_owned(),
            incarnation: id('a'),
        }
    }

    fn resource_spec() -> IndexResourceSpec {
        IndexResourceSpec {
            format_version: u32::from(LODESTONE_API_INDEX_RESOURCE_SPEC_FORMAT_VERSION),
            revision: U64 { value: "1".to_owned() },
            initial_shards: 1,
            target_shard_bytes: U64 { value: "1".to_owned() },
            durability: ResourceDurabilityClass::RegionalObject,
            storage_class: ResourceStorageClass::Standard,
            ram_admission_weight: 1,
            ram_hard_max_bytes: U64 { value: "1".to_owned() },
            nvme_admission_weight: 1,
            nvme_hard_max_bytes: U64 { value: "1".to_owned() },
            refresh_objective_millis: 1,
            memtable_hard_max_bytes: U64 { value: "1".to_owned() },
            seal_parallelism: 1,
            merge_parallelism: 1,
            vector: lodestone_wire::VectorResourcePolicy {
                quality: lodestone_wire::VectorQualityObjective::Balanced,
                indexing: lodestone_wire::VectorIndexingObjective::Balanced,
                residency: None,
                calibration: None,
            },
            lifecycle_binding: None,
            encryption_required: true,
            resolver_capability: None,
        }
    }

    fn valid_chain() -> Vec<ProvisionRetainedAppendTransition> {
        let resource = resource();
        vec![
            ProvisionRetainedAppendTransition::PreflightVerified,
            ProvisionRetainedAppendTransition::CatalogPrepared {
                locator: locator("catalog-1"),
                semantic_digest: digest('b'),
                expires_at: U64 { value: "10".to_owned() },
                resource_spec: resource_spec(),
            },
            ProvisionRetainedAppendTransition::CatalogApplied {
                locator: locator("catalog-1"),
                resource: resource.clone(),
            },
            ProvisionRetainedAppendTransition::IndexActive {
                resource: resource.clone(),
                index_revision: U64 { value: "1".to_owned() },
            },
            ProvisionRetainedAppendTransition::RolePrepared {
                locator: locator("role-1"),
                semantic_digest: digest('c'),
                expires_at: U64 { value: "20".to_owned() },
                predecessor_role_version: U64 { value: "4".to_owned() },
                successor_role_version: U64 { value: "5".to_owned() },
            },
            ProvisionRetainedAppendTransition::RoleApplied {
                locator: locator("role-1"),
                resolution: RoleAppendApplyResolution::Resolver,
                target_policy_revision: U64 { value: "9".to_owned() },
            },
            ProvisionRetainedAppendTransition::RoleVerified {
                resource: resource.clone(),
                role_version: U64 { value: "6".to_owned() },
                target_policy_revision: Some(U64 { value: "9".to_owned() }),
            },
            ProvisionRetainedAppendTransition::DiscoveryVerified {
                resource: resource.clone(),
                observed_policy_revision: U64 { value: "10".to_owned() },
            },
            ProvisionRetainedAppendTransition::Completed {
                resource,
                role_version: U64 { value: "6".to_owned() },
                observed_policy_revision: U64 { value: "10".to_owned() },
            },
        ]
    }

    #[test]
    fn reducer_accepts_monotonic_completed_chain() -> Result<(), WorkflowJournalError> {
        let state = ProvisionRetainedAppendState::replay(&valid_chain())?;
        assert!(state.completed());
        assert_eq!(state.resource(), Some(&resource()));
        Ok(())
    }

    #[test]
    fn intent_and_transitions_reject_zero_authority_values() {
        let intent = ProvisionRetainedAppendIntent {
            format_version: PROVISION_RETAINED_APPEND_FORMAT_VERSION,
            cluster_id: id('0'),
            collection_id: id('a'),
            admin_profile_sha256: [1; 32],
            consumer_profile_sha256: [2; 32],
            trust_sha256: [3; 32],
            definition_sha256: [4; 32],
            index_name: "events".to_owned(),
            role_name: "consumer".to_owned(),
        };
        assert_eq!(intent.validate(), Err(WorkflowJournalError::InvalidIntent));

        let mut intent = intent;
        intent.cluster_id = id('a');
        intent.definition_sha256 = [0; 32];
        assert_eq!(intent.validate(), Err(WorkflowJournalError::InvalidIntent));

        let transition = ProvisionRetainedAppendTransition::CatalogPrepared {
            locator: locator("catalog-1"),
            semantic_digest: digest('0'),
            expires_at: U64 { value: "10".to_owned() },
            resource_spec: resource_spec(),
        };
        assert_eq!(transition.validate(), Err(WorkflowJournalError::InvalidTransition));
    }

    #[test]
    fn reducer_rejects_out_of_order_or_post_terminal_records() {
        assert_eq!(
            ProvisionRetainedAppendState::replay(&[
                ProvisionRetainedAppendTransition::PreflightVerified,
                ProvisionRetainedAppendTransition::IndexActive {
                    resource: resource(),
                    index_revision: U64 { value: "1".to_owned() },
                }
            ]),
            Err(WorkflowJournalError::Corrupt)
        );
        let mut transitions = valid_chain();
        transitions.push(ProvisionRetainedAppendTransition::PreflightVerified);
        assert_eq!(
            ProvisionRetainedAppendState::replay(&transitions),
            Err(WorkflowJournalError::Corrupt)
        );
    }

    #[test]
    fn reducer_retires_only_the_exact_role_locator() {
        let mut transitions = valid_chain();
        transitions.truncate(5);
        transitions.push(ProvisionRetainedAppendTransition::RoleAttemptClosed {
            locator: locator("other"),
            closure: RoleAppendAttemptClosure::ExpiredUnknown,
        });
        assert_eq!(
            ProvisionRetainedAppendState::replay(&transitions),
            Err(WorkflowJournalError::Corrupt)
        );
    }

    #[test]
    fn reducer_accepts_a_later_role_version_that_retains_the_scope() -> Result<(), WorkflowJournalError> {
        let mut transitions = valid_chain();
        let Some(ProvisionRetainedAppendTransition::Completed { role_version, .. }) = transitions.last_mut() else {
            return Err(WorkflowJournalError::Corrupt);
        };
        *role_version = U64 { value: "7".to_owned() };
        let state = ProvisionRetainedAppendState::replay(&transitions)?;
        assert_eq!(state.role_version(), Some(&U64 { value: "7".to_owned() }));
        Ok(())
    }
}