lodestone-proto 0.0.16

//! Strict native administration request shapes.
//!
//! The authenticated collection is intentionally absent. A listener supplies it
//! from request context before translating this leaf into the metastore's
//! administration command.

use lodestone_types::{
    AdministrationActionId, AdministrationRevision, AliasRevision, CollectionIndexRef, ComponentTemplateId,
    ComponentTemplateRevision, DataStreamRevision, IndexName, IndexResourceSpecV1, IndexTemplateId,
    IndexTemplateRevision, IngestPipelineId, IngestPipelineRevision, LifecyclePolicyId, LifecyclePolicyRevision,
    TaskCheckpointGeneration, TaskId,
};
use serde::{Deserialize, Serialize};

use crate::RequestId;

pub const NATIVE_ADMINISTRATION_FORMAT_VERSION: u16 = 1;
pub const MAX_NATIVE_ADMINISTRATION_ACTIONS: usize = 1_024;
pub const MAX_NATIVE_ADMINISTRATION_PAYLOAD_BYTES: usize = 256 * 1024;

pub fn native_administration_action_id(
    request_id: &RequestId,
) -> Result<AdministrationActionId, lodestone_types::IdentityError> {
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"lodestone/native-administration-action/v1\0");
    hasher.update(request_id.as_str().as_bytes());
    AdministrationActionId::new(*hasher.finalize().as_bytes())
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NativeAdministrationRequestV1 {
    #[serde(rename = "command")]
    command: String,
    pub format_version: u16,
    pub request_id: RequestId,
    pub action_id: AdministrationActionId,
    pub operation: NativeAdministrationOperationV1,
}

impl NativeAdministrationRequestV1 {
    #[must_use]
    pub fn new(
        request_id: RequestId,
        action_id: AdministrationActionId,
        operation: NativeAdministrationOperationV1,
    ) -> Self {
        Self {
            command: "administration".to_owned(),
            format_version: NATIVE_ADMINISTRATION_FORMAT_VERSION,
            request_id,
            action_id,
            operation,
        }
    }

    pub fn validate_semantics(&self) -> Result<(), NativeAdministrationContractError> {
        if self.command != "administration" {
            return Err(NativeAdministrationContractError::WrongCommand);
        }
        if self.format_version != NATIVE_ADMINISTRATION_FORMAT_VERSION {
            return Err(NativeAdministrationContractError::UnsupportedVersion(
                self.format_version,
            ));
        }
        if let NativeAdministrationOperationV1::Mutate { mutation, .. } = &self.operation {
            mutation.validate_semantics()?;
        }
        if let NativeAdministrationOperationV1::SimulatePipeline {
            name,
            expected,
            definition,
            documents,
        } = &self.operation
            && (name.is_some() == definition.is_some()
                || documents.is_empty()
                || documents.len() > 32
                || documents
                    .iter()
                    .any(|document| document.is_empty() || document.len() > 1024 * 1024)
                || (expected.is_some() && name.is_none()))
        {
            return Err(NativeAdministrationContractError::InvalidPipelineSimulation);
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum NativeAdministrationOperationV1 {
    Mutate {
        expected_revision: AdministrationRevision,
        mutation: NativeAdministrationMutationV1,
    },
    Inspect,
    ResolveName {
        name: IndexName,
    },
    InspectTask {
        task_id: TaskId,
    },
    CancelTask {
        task_id: TaskId,
        expected_generation: TaskCheckpointGeneration,
    },
    SimulatePipeline {
        name: Option<String>,
        expected: Option<NativeIngestPipelineBindingV1>,
        definition: Option<serde_json::Value>,
        documents: Vec<Vec<u8>>,
    },
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NativeAliasTargetV1 {
    pub collection_index_ref: CollectionIndexRef,
    #[serde(default)]
    pub write_index: bool,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum NativeAliasActionV1 {
    Put {
        name: IndexName,
        #[serde(default)]
        expected_revision: Option<AliasRevision>,
        targets: Vec<NativeAliasTargetV1>,
        #[serde(default)]
        filter: Option<serde_json::Value>,
        #[serde(default)]
        index_routing: Option<String>,
        #[serde(default)]
        search_routing: Option<String>,
    },
    Delete {
        name: IndexName,
        expected_revision: AliasRevision,
    },
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NativeComponentTemplateBindingV1 {
    pub template_id: ComponentTemplateId,
    pub revision: ComponentTemplateRevision,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NativeIndexTemplateBindingV1 {
    pub template_id: IndexTemplateId,
    pub revision: IndexTemplateRevision,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NativeIngestPipelineBindingV1 {
    pub pipeline_id: IngestPipelineId,
    pub revision: IngestPipelineRevision,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NativeLifecyclePolicyBindingV1 {
    pub policy_id: LifecyclePolicyId,
    pub revision: LifecyclePolicyRevision,
}

#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NativeRolloverThresholdsV1 {
    pub max_primary_docs: Option<u64>,
    pub max_primary_bytes: Option<u64>,
    pub max_age_millis: Option<u64>,
}

#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NativeRetentionPolicyV1 {
    pub document_max_age_millis: Option<u64>,
    pub backing_index_max_age_millis: Option<u64>,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum NativeAdministrationMutationV1 {
    ApplyAliasActions {
        actions: Vec<NativeAliasActionV1>,
    },
    PutComponentTemplate {
        name: String,
        expected: Option<NativeComponentTemplateBindingV1>,
        mapping: serde_json::Value,
        settings: serde_json::Value,
        resource_spec: Option<Box<IndexResourceSpecV1>>,
    },
    DeleteComponentTemplate {
        name: String,
        expected: NativeComponentTemplateBindingV1,
    },
    PutIndexTemplate {
        name: String,
        expected: Option<NativeIndexTemplateBindingV1>,
        priority: i32,
        data_stream: bool,
        index_patterns: Vec<String>,
        components: Vec<NativeComponentTemplateBindingV1>,
        mapping: serde_json::Value,
        settings: serde_json::Value,
        resource_spec: Option<Box<IndexResourceSpecV1>>,
    },
    DeleteIndexTemplate {
        name: String,
        expected: NativeIndexTemplateBindingV1,
    },
    PutIngestPipeline {
        name: String,
        expected: Option<NativeIngestPipelineBindingV1>,
        definition: serde_json::Value,
    },
    DeleteIngestPipeline {
        name: String,
        expected: NativeIngestPipelineBindingV1,
    },
    PutLifecyclePolicy {
        name: String,
        expected: Option<NativeLifecyclePolicyBindingV1>,
        rollover: NativeRolloverThresholdsV1,
        retention: NativeRetentionPolicyV1,
        merge_hint: bool,
        freeze_hint: bool,
        delete_expired: bool,
    },
    DeleteLifecyclePolicy {
        name: String,
        expected: NativeLifecyclePolicyBindingV1,
    },
    CreateDataStream {
        name: IndexName,
        timestamp_field: String,
        index_template: NativeIndexTemplateBindingV1,
        lifecycle_policy: Option<NativeLifecyclePolicyBindingV1>,
    },
    DeleteDataStream {
        name: IndexName,
        expected_revision: DataStreamRevision,
    },
    RolloverDataStream {
        name: IndexName,
        expected_revision: DataStreamRevision,
    },
}

impl NativeAdministrationMutationV1 {
    fn validate_semantics(&self) -> Result<(), NativeAdministrationContractError> {
        if let Self::ApplyAliasActions { actions } = self
            && (actions.is_empty() || actions.len() > MAX_NATIVE_ADMINISTRATION_ACTIONS)
        {
            return Err(NativeAdministrationContractError::InvalidActionCount(actions.len()));
        }
        let bytes = serde_json::to_vec(self).map_err(NativeAdministrationContractError::Encode)?;
        if bytes.len() > MAX_NATIVE_ADMINISTRATION_PAYLOAD_BYTES {
            return Err(NativeAdministrationContractError::PayloadTooLarge(bytes.len()));
        }
        Ok(())
    }
}

#[derive(Debug)]
pub enum NativeAdministrationContractError {
    Encode(serde_json::Error),
    WrongCommand,
    UnsupportedVersion(u16),
    InvalidActionCount(usize),
    PayloadTooLarge(usize),
    InvalidPipelineSimulation,
}

impl std::fmt::Display for NativeAdministrationContractError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Encode(error) => write!(formatter, "administration contract encoding failed: {error}"),
            Self::WrongCommand => formatter.write_str("administration decoder received another command"),
            Self::UnsupportedVersion(version) => write!(formatter, "unsupported administration version {version}"),
            Self::InvalidActionCount(count) => write!(formatter, "invalid alias action count {count}"),
            Self::PayloadTooLarge(bytes) => write!(formatter, "administration payload is {bytes} bytes"),
            Self::InvalidPipelineSimulation => {
                formatter.write_str("invalid pipeline simulation selection or document bound")
            }
        }
    }
}

impl std::error::Error for NativeAdministrationContractError {}

#[cfg(test)]
mod tests {
    use super::*;

    fn action_json(byte: u8) -> String {
        serde_json::to_string(&[byte; 32]).unwrap_or_default()
    }

    #[test]
    fn semantic_contract_refuses_unknown_fields_and_empty_alias_batch() -> Result<(), Box<dyn std::error::Error>> {
        let unknown = br#"{"command":"administration","format_version":1,"request_id":"r","action_id":[1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1],"operation":{"kind":"inspect"},"extra":1}"#;
        assert!(serde_json::from_slice::<NativeAdministrationRequestV1>(unknown).is_err());
        let empty = br#"{"command":"administration","format_version":1,"request_id":"r","action_id":[1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1],"operation":{"kind":"mutate","expected_revision":0,"mutation":{"kind":"apply_alias_actions","actions":[]}}}"#;
        let request: NativeAdministrationRequestV1 = serde_json::from_slice(empty)?;
        assert!(matches!(
            request.validate_semantics(),
            Err(NativeAdministrationContractError::InvalidActionCount(0))
        ));
        Ok(())
    }

    #[test]
    fn semantic_contract_refuses_newer_zero_malformed_and_oversized_values() -> Result<(), Box<dyn std::error::Error>> {
        let newer = format!(
            r#"{{"command":"administration","format_version":2,"request_id":"r","action_id":{},"operation":{{"kind":"inspect"}}}}"#,
            action_json(1)
        );
        let newer: NativeAdministrationRequestV1 = serde_json::from_str(&newer)?;
        assert!(matches!(
            newer.validate_semantics(),
            Err(NativeAdministrationContractError::UnsupportedVersion(2))
        ));
        let zero = format!(
            r#"{{"command":"administration","format_version":1,"request_id":"r","action_id":{},"operation":{{"kind":"inspect"}}}}"#,
            action_json(0)
        );
        assert!(serde_json::from_str::<NativeAdministrationRequestV1>(&zero).is_err());
        assert!(serde_json::from_slice::<NativeAdministrationRequestV1>(br#"{"command":"administration""#).is_err());
        let oversized = serde_json::json!({
            "command": "administration", "format_version": 1, "request_id": "r", "action_id": vec![1; 32],
            "operation": {"kind": "mutate", "expected_revision": 0, "mutation": {
                "kind": "put_component_template", "name": "component", "expected": null,
                "mapping": {"payload": "x".repeat(MAX_NATIVE_ADMINISTRATION_PAYLOAD_BYTES)}, "settings": {}
            }}
        });
        let request: NativeAdministrationRequestV1 = serde_json::from_value(oversized)?;
        assert!(matches!(
            request.validate_semantics(),
            Err(NativeAdministrationContractError::PayloadTooLarge(_))
        ));
        Ok(())
    }
}