dynoxide-rs 3.0.0

use crate::actions::{TableDescription, build_table_description};
use crate::errors::{DynoxideError, Result};
use crate::storage_backend::{BackendError, StorageBackend};
use crate::streams;
use crate::types::{
    AttributeDefinition, GlobalSecondaryIndex, KeySchemaElement, KeyType, LocalSecondaryIndex,
    Projection, ProjectionType, ProvisionedThroughput, VectorIndex,
};
use serde::{Deserialize, Serialize};
use web_time::{SystemTime, UNIX_EPOCH};

/// Internal raw deserialization struct: uses serde_json::Value for fields
/// that participate in DynamoDB's multi-field constraint validation.
#[derive(Debug, Default, Deserialize)]
struct RawRequest {
    #[serde(rename = "TableName", default)]
    table_name: Option<String>,
    #[serde(rename = "KeySchema", default)]
    key_schema: Option<serde_json::Value>,
    #[serde(rename = "AttributeDefinitions", default)]
    attribute_definitions: Option<serde_json::Value>,
    #[serde(rename = "GlobalSecondaryIndexes", default)]
    global_secondary_indexes: Option<serde_json::Value>,
    #[serde(rename = "LocalSecondaryIndexes", default)]
    local_secondary_indexes: Option<serde_json::Value>,
    #[serde(rename = "VectorIndexes", default)]
    vector_indexes: Option<serde_json::Value>,
    #[serde(rename = "BillingMode", default)]
    billing_mode: Option<String>,
    #[serde(rename = "ProvisionedThroughput", default)]
    provisioned_throughput: Option<serde_json::Value>,
    #[serde(rename = "StreamSpecification", default)]
    stream_specification: Option<StreamSpecification>,
    #[serde(rename = "SSESpecification", default)]
    sse_specification: Option<crate::types::SseSpecification>,
    #[serde(rename = "TableClass", default)]
    table_class: Option<String>,
    #[serde(rename = "Tags", default)]
    tags: Option<Vec<crate::types::Tag>>,
    #[serde(rename = "DeletionProtectionEnabled", default)]
    deletion_protection_enabled: Option<bool>,
    #[serde(rename = "OnDemandThroughput", default)]
    on_demand_throughput: Option<crate::types::OnDemandThroughput>,
}

/// Public request type: fully validated, typed fields.
/// Can be constructed directly (programmatic use) or deserialized from JSON.
#[derive(Debug, Default)]
pub struct CreateTableRequest {
    pub table_name: String,
    pub key_schema: Vec<KeySchemaElement>,
    pub attribute_definitions: Vec<AttributeDefinition>,
    pub global_secondary_indexes: Option<Vec<GlobalSecondaryIndex>>,
    pub local_secondary_indexes: Option<Vec<LocalSecondaryIndex>>,
    pub vector_indexes: Option<Vec<VectorIndex>>,
    pub billing_mode: Option<String>,
    pub provisioned_throughput: Option<ProvisionedThroughput>,
    pub stream_specification: Option<StreamSpecification>,
    pub sse_specification: Option<crate::types::SseSpecification>,
    pub table_class: Option<String>,
    pub tags: Option<Vec<crate::types::Tag>>,
    pub deletion_protection_enabled: Option<bool>,
    pub on_demand_throughput: Option<crate::types::OnDemandThroughput>,
}

/// Custom Deserialize that does loose JSON parsing first, validates, then builds typed fields.
/// Validation errors use "VALIDATION:" prefix so server.rs converts them to ValidationException.
impl<'de> serde::Deserialize<'de> for CreateTableRequest {
    fn deserialize<D: serde::Deserializer<'de>>(
        deserializer: D,
    ) -> std::result::Result<Self, D::Error> {
        let raw = RawRequest::deserialize(deserializer)?;
        match validate_raw_and_build(raw) {
            Ok(req) => Ok(req),
            Err(msg) => Err(serde::de::Error::custom(format!("VALIDATION:{}", msg))),
        }
    }
}

#[derive(Debug, Default, Deserialize)]
pub struct StreamSpecification {
    #[serde(rename = "StreamEnabled", alias = "stream_enabled")]
    pub stream_enabled: bool,
    #[serde(rename = "StreamViewType", alias = "stream_view_type", default)]
    pub stream_view_type: Option<String>,
}

#[derive(Debug, Default, Serialize)]
pub struct CreateTableResponse {
    #[serde(rename = "TableDescription")]
    pub table_description: TableDescription,
}

pub async fn execute<S: StorageBackend>(
    storage: &S,
    mut request: CreateTableRequest,
) -> Result<CreateTableResponse> {
    // An OnDemandThroughput object with no members is equivalent to omitting
    // it: real DynamoDB accepts it at creation and stores nothing (eu-west-2
    // capture, 2026-07-24), so normalise before any gate looks at it.
    if request.on_demand_throughput.as_ref().is_some_and(|odt| {
        odt.max_read_request_units.is_none() && odt.max_write_request_units.is_none()
    }) {
        request.on_demand_throughput = None;
    }

    // Structural validation (runs for both programmatic and JSON paths)
    validate_typed_request(&request)?;

    if let Some(ref bm) = request.billing_mode
        && bm != "PROVISIONED"
        && bm != "PAY_PER_REQUEST"
    {
        return Err(DynoxideError::ValidationException(format!(
            "1 validation error detected: Value '{bm}' at 'billingMode' failed to satisfy \
                 constraint: Member must satisfy enum value set: \
                 [PROVISIONED, PAY_PER_REQUEST]"
        )));
    }

    if let Some(ref tc) = request.table_class
        && tc != "STANDARD"
        && tc != "STANDARD_INFREQUENT_ACCESS"
    {
        return Err(DynoxideError::ValidationException(format!(
            "1 validation error detected: Value '{tc}' at 'tableClass' failed to satisfy \
                 constraint: Member must satisfy enum value set: \
                 [STANDARD, STANDARD_INFREQUENT_ACCESS]"
        )));
    }

    // OnDemandThroughput is only valid alongside PAY_PER_REQUEST billing, and
    // its members must be at least 1 at creation (-1 is an UpdateTable-only
    // removal marker). Both messages captured from real DynamoDB (eu-west-2,
    // 2026-07-24); the gate names the first present member, read checked
    // first, and the create wording drops "the" and carries a full stop.
    if let Some(ref odt) = request.on_demand_throughput {
        let members = [
            ("MaxReadRequestUnits", odt.max_read_request_units),
            ("MaxWriteRequestUnits", odt.max_write_request_units),
        ];
        if let Some((member, _)) = members.iter().find(|(_, v)| v.is_some()) {
            let billing_mode_str = request.billing_mode.as_deref().unwrap_or("PROVISIONED");
            if billing_mode_str == "PROVISIONED" {
                return Err(DynoxideError::ValidationException(format!(
                    "One or more parameter values were invalid: {member} for \
                     OnDemandThroughput cannot be specified when table BillingMode \
                     is PROVISIONED."
                )));
            }
        }
        for (member, value) in members {
            if value.is_some_and(|v| v < 1) {
                return Err(DynoxideError::ValidationException(format!(
                    "One or more parameter values were invalid: Requested {member} for \
                     OnDemandThroughput for table is outside of valid range"
                )));
            }
        }
    }

    if storage.table_exists(&request.table_name).await? {
        return Err(DynoxideError::ResourceInUseException(format!(
            "Table already exists: {}",
            request.table_name
        )));
    }

    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs() as i64;

    let key_schema_json = serde_json::to_string(&request.key_schema)
        .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
    let attr_defs_json = serde_json::to_string(&request.attribute_definitions)
        .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
    let gsi_json = request
        .global_secondary_indexes
        .as_ref()
        .map(serde_json::to_string)
        .transpose()
        .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
    let lsi_json = request
        .local_secondary_indexes
        .as_ref()
        .map(serde_json::to_string)
        .transpose()
        .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
    let vector_json = request
        .vector_indexes
        .as_ref()
        .map(serde_json::to_string)
        .transpose()
        .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
    let pt_json = request
        .provisioned_throughput
        .as_ref()
        .map(serde_json::to_string)
        .transpose()
        .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
    // Normalise the SSE spec so a DescribeTable round-trip matches AWS: when
    // encryption is enabled without an explicit type/key, AWS reports SSEType=KMS
    // and a KMS key ARN (the AWS-managed `aws/dynamodb` key). Persisting the
    // synthesised key id keeps the reported ARN stable across DescribeTable calls.
    let normalized_sse = request.sse_specification.as_ref().map(|spec| {
        if spec.enabled == Some(true) {
            crate::types::SseSpecification {
                enabled: Some(true),
                sse_type: spec.sse_type.clone().or_else(|| Some("KMS".to_string())),
                kms_master_key_id: spec.kms_master_key_id.clone().or_else(|| {
                    Some(crate::streams::kms_key_arn(
                        &uuid::Uuid::new_v4().to_string(),
                    ))
                }),
            }
        } else {
            spec.clone()
        }
    });
    let sse_json = normalized_sse
        .as_ref()
        .map(serde_json::to_string)
        .transpose()
        .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
    let on_demand_json = request
        .on_demand_throughput
        .as_ref()
        .map(serde_json::to_string)
        .transpose()
        .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
    let deletion_protection = request.deletion_protection_enabled.unwrap_or(false);

    // Refuse capability gaps before creating anything. A refusal after
    // `insert_table_metadata` would leave the table half-created: the client
    // sees an error, retries, and finds ResourceInUseException where the real
    // answer is that the backend cannot honour the request at all.
    if let Some(ref spec) = request.stream_specification
        && spec.stream_enabled
        && !storage.supports_streams()
    {
        return Err(BackendError::Unsupported {
            capability: "streams",
        }
        .into());
    }
    if request.tags.as_ref().is_some_and(|tags| !tags.is_empty()) && !storage.supports_tags() {
        return Err(BackendError::Unsupported { capability: "tags" }.into());
    }

    let billing_mode_str = request.billing_mode.as_deref().unwrap_or("PROVISIONED");
    storage
        .insert_table_metadata(&crate::storage::CreateTableMetadata {
            table_name: &request.table_name,
            key_schema: &key_schema_json,
            attribute_definitions: &attr_defs_json,
            gsi_definitions: gsi_json.as_deref(),
            lsi_definitions: lsi_json.as_deref(),
            provisioned_throughput: pt_json.as_deref(),
            created_at: now,
            sse_specification: sse_json.as_deref(),
            table_class: request.table_class.as_deref(),
            deletion_protection_enabled: deletion_protection,
            billing_mode: Some(billing_mode_str),
            on_demand_throughput: on_demand_json.as_deref(),
            vector_index_definitions: vector_json.as_deref(),
        })
        .await?;

    storage.create_data_table(&request.table_name).await?;

    if let Some(ref gsis) = request.global_secondary_indexes {
        for gsi in gsis {
            storage
                .create_gsi_table(&request.table_name, &gsi.index_name)
                .await?;
        }
    }

    if let Some(ref lsis) = request.local_secondary_indexes {
        for lsi in lsis {
            storage
                .create_lsi_table(&request.table_name, &lsi.index_name)
                .await?;
        }
    }

    if let Some(ref vixs) = request.vector_indexes {
        for vix in vixs {
            storage
                .create_vector_table(&request.table_name, &vix.index_name)
                .await?;
        }
    }

    if let Some(ref spec) = request.stream_specification
        && spec.stream_enabled
    {
        let view_type = spec
            .stream_view_type
            .as_deref()
            .unwrap_or("NEW_AND_OLD_IMAGES");
        let label = streams::generate_stream_label(storage.clock());
        storage
            .enable_stream(&request.table_name, view_type, &label)
            .await?;
    }

    if let Some(ref tags) = request.tags
        && !tags.is_empty()
    {
        storage.set_tags(&request.table_name, tags).await?;
    }

    let meta = storage
        .get_table_metadata(&request.table_name)
        .await?
        .ok_or_else(|| {
            DynoxideError::InternalServerError("Table metadata not found after creation".into())
        })?;

    // An index created with its table is never armed: measured in eu-west-2 on
    // 2026-08-21 it reaches ACTIVE in the same DescribeTable poll as the table,
    // so there is no window to report and an empty phase set says exactly that.
    let mut desc = build_table_description(
        &meta,
        Some(0),
        Some(0),
        &crate::actions::vector_lifecycle::VectorIndexPhases::default(),
    );
    // CreateTable response shows CREATING status (table is usable immediately
    // but DynamoDB API contract says newly-created tables start as CREATING)
    desc.table_status = "CREATING".to_string();

    // Override billing mode fields based on the actual request
    let billing_mode_str = request.billing_mode.as_deref().unwrap_or("PROVISIONED");
    if billing_mode_str == "PROVISIONED" {
        desc.billing_mode_summary = None;
        desc.table_throughput_mode_summary = None;
    } else if billing_mode_str == "PAY_PER_REQUEST" {
        desc.billing_mode_summary = Some(crate::actions::BillingModeSummary {
            billing_mode: "PAY_PER_REQUEST".to_string(),
            last_update_to_pay_per_request_date_time: None,
        });
        desc.table_throughput_mode_summary = Some(crate::actions::TableThroughputModeSummary {
            table_throughput_mode: "PAY_PER_REQUEST".to_string(),
            last_update_to_pay_per_request_date_time: None,
        });
        // Ensure provisioned throughput shows zeros for PAY_PER_REQUEST
        desc.provisioned_throughput = Some(crate::actions::TableProvisionedThroughputDescription {
            read_capacity_units: 0,
            write_capacity_units: 0,
            number_of_decreases_today: 0,
            last_increase_date_time: None,
            last_decrease_date_time: None,
        });
    }

    // Set all GSI statuses to CREATING for newly created tables
    if let Some(ref mut gsis) = desc.global_secondary_indexes {
        for gsi in gsis {
            gsi.index_status = "CREATING".to_string();
        }
    }

    // Vector indexes likewise report CREATING in the CreateTable response;
    // DescribeTable then reports ACTIVE with Backfilling never present on
    // this path (captured from real DynamoDB, eu-west-2, 2026-08-11).
    if let Some(ref mut vixs) = desc.vector_indexes {
        for vix in vixs {
            vix.index_status = "CREATING".to_string();
        }
    }

    // Remove DeletionProtectionEnabled from response if not explicitly set
    // (DynamoDB doesn't include it in basic CreateTable response)
    if request.deletion_protection_enabled.is_none() {
        desc.deletion_protection_enabled = None;
    }

    Ok(CreateTableResponse {
        table_description: desc,
    })
}

/// Convert a String error to DynoxideError::ValidationException.
fn ve(msg: String) -> DynoxideError {
    DynoxideError::ValidationException(msg)
}

/// Validate a programmatically-constructed request (used when not deserialised from JSON).
///
/// The validation order matches DynamoDB's actual behaviour (as verified by the Dynalite
/// conformance suite):
///
/// 1. Table name (missing, length, pattern)
/// 2. BillingMode + ProvisionedThroughput consistency
/// 3. ProvisionedThroughput out-of-bounds
/// 4. Missing ProvisionedThroughput (default PROVISIONED billing)
/// 5. Key attribute definition checks ("Invalid KeySchema" / detailed missing-attr message)
/// 6. Key schema structure (duplicate names, wrong types)
/// 7. Empty LSI/GSI/VectorIndexes lists
/// 8. LSI/GSI structural validation (key schema, projections, duplicates, limits)
/// 9. Vector index validation (billing-mode gate, per-index structure,
///    duplicates, count limit, conflicting dimensions)
/// 10. Cross-index duplicate names (LSI/GSI/vector share one namespace)
/// 11. Attribute definition count mismatch (SearchSchema attributes join the
///     used set)
/// 12. StreamSpecification consistency (a disabled stream must not set a view type)
fn validate_typed_request(request: &CreateTableRequest) -> Result<()> {
    if request.table_name.is_empty() {
        return Err(DynoxideError::ValidationException(
            "The parameter 'TableName' is required but was not present in the request".to_string(),
        ));
    }
    if request.table_name.len() < 3 || request.table_name.len() > 255 {
        return Err(DynoxideError::ValidationException(
            "TableName must be at least 3 characters long and at most 255 characters long"
                .to_string(),
        ));
    }

    // Table name pattern
    if !request
        .table_name
        .chars()
        .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-' || c == '.')
    {
        return Err(DynoxideError::ValidationException(format!(
            "1 validation error detected: Value '{}' at 'tableName' failed to satisfy constraint: \
             Member must satisfy regular expression pattern: [a-zA-Z0-9_.-]+",
            request.table_name
        )));
    }

    // BillingMode + ProvisionedThroughput consistency
    let billing_mode_str = request.billing_mode.as_deref().unwrap_or("PROVISIONED");
    if billing_mode_str == "PAY_PER_REQUEST" && request.provisioned_throughput.is_some() {
        return Err(DynoxideError::ValidationException(
            "One or more parameter values were invalid: Neither ReadCapacityUnits nor \
             WriteCapacityUnits can be specified when BillingMode is PAY_PER_REQUEST"
                .to_string(),
        ));
    }

    // ProvisionedThroughput out-of-bounds
    if let Some(ref pt) = request.provisioned_throughput {
        const MAX_THROUGHPUT: i64 = 1_000_000_000_000;
        let rcu = pt.read_capacity_units.unwrap_or(0);
        let wcu = pt.write_capacity_units.unwrap_or(0);
        if rcu > MAX_THROUGHPUT {
            return Err(DynoxideError::ValidationException(format!(
                "Given value {} for ReadCapacityUnits is out of bounds",
                rcu
            )));
        }
        if wcu > MAX_THROUGHPUT {
            return Err(DynoxideError::ValidationException(format!(
                "Given value {} for WriteCapacityUnits is out of bounds",
                wcu
            )));
        }
    }

    // Missing ProvisionedThroughput when billing mode is explicitly PROVISIONED.
    // For the programmatic API, when BillingMode is not specified we default to
    // PAY_PER_REQUEST for convenience. The HTTP/JSON path (validate_raw_and_build)
    // applies the stricter DynamoDB default of PROVISIONED.
    if request.billing_mode.is_some()
        && billing_mode_str == "PROVISIONED"
        && request.provisioned_throughput.is_none()
    {
        return Err(DynoxideError::ValidationException(
            "One or more parameter values were invalid: ReadCapacityUnits and \
             WriteCapacityUnits must both be specified when BillingMode is PROVISIONED"
                .to_string(),
        ));
    }

    // Key attribute definition checks (before key schema structure)
    validate_key_attrs_in_defs(&request.key_schema, &request.attribute_definitions).map_err(ve)?;

    // Key schema structure
    validate_key_schema_structure(&request.key_schema).map_err(ve)?;

    // Empty LSI/GSI lists (before structural validation and attr count)
    if let Some(ref lsis) = request.local_secondary_indexes
        && lsis.is_empty()
    {
        return Err(ve(
            "One or more parameter values were invalid: List of LocalSecondaryIndexes is empty"
                .to_string(),
        ));
    }
    if let Some(ref gsis) = request.global_secondary_indexes
        && gsis.is_empty()
    {
        return Err(ve(
            "One or more parameter values were invalid: List of GlobalSecondaryIndexes is empty"
                .to_string(),
        ));
    }
    // An empty VectorIndexes list is rejected, not normalised to absent.
    // Captured from real DynamoDB (eu-west-2 and us-east-1, 2026-08-12).
    if let Some(ref vixs) = request.vector_indexes
        && vixs.is_empty()
    {
        return Err(ve(
            "One or more parameter values were invalid: List of VectorIndexes is empty".to_string(),
        ));
    }

    // LSI structural validation
    if let Some(ref lsis) = request.local_secondary_indexes {
        validate_lsi_list(lsis, &request.key_schema, &request.attribute_definitions).map_err(ve)?;
    }

    // GSI structural validation
    if let Some(ref gsis) = request.global_secondary_indexes {
        let bm = request.billing_mode.as_deref().unwrap_or("PROVISIONED");
        validate_gsi_list(gsis, &request.attribute_definitions, bm).map_err(ve)?;
    }

    // Vector index validation (the empty list was rejected above)
    if let Some(ref vixs) = request.vector_indexes {
        let bm = request.billing_mode.as_deref().unwrap_or("PROVISIONED");
        validate_vector_index_list(vixs, &request.attribute_definitions, bm)?;
    }

    // Cross-index duplicate names (checked before attr def count)
    check_cross_index_duplicates(
        &request.local_secondary_indexes,
        &request.global_secondary_indexes,
        &request.vector_indexes,
    )
    .map_err(ve)?;

    // Attribute definition count (last of the key/index checks). SearchSchema
    // attributes count towards the used set, so declaring them does not trip
    // the exact-match rule.
    validate_attr_def_count(
        &request.key_schema,
        &request.attribute_definitions,
        &request.local_secondary_indexes,
        &request.global_secondary_indexes,
        &request.vector_indexes,
    )
    .map_err(ve)?;

    // StreamSpecification consistency: a disabled stream must not carry a view
    // type. Real DynamoDB rejects the combination, because a view type only has
    // meaning when the stream is enabled.
    if let Some(ref spec) = request.stream_specification
        && !spec.stream_enabled
        && spec.stream_view_type.is_some()
    {
        return Err(DynoxideError::ValidationException(
            "One or more parameter values were invalid: Table is being created with a stream \
                 disabled, UpdateViewType should not be specified"
                .to_string(),
        ));
    }

    Ok(())
}

/// List-level vector index validation, shared by the JSON and programmatic
/// paths via `validate_typed_request`. All operation-layer messages here are
/// captured from real DynamoDB (eu-west-2, 2026-08-11); the duplicate-name
/// form is the classic secondary-index message, captured for vector indexes
/// too, both same-family and cross-family (eu-west-2 and us-east-1,
/// 2026-08-12).
///
/// Returns the crate `Result` rather than the `Result<(), String>` of the
/// `validate_lsi_list`/`validate_gsi_list` siblings: `validate_vector_index`
/// raises typed `DynoxideError`s at source, so there is no string layer to
/// map through here.
fn validate_vector_index_list(
    vixs: &[VectorIndex],
    defs: &[AttributeDefinition],
    bm: &str,
) -> Result<()> {
    if bm != "PAY_PER_REQUEST" {
        return Err(ve(
            "One or more parameter values were invalid: Vector indexes are only supported \
             for PAY_PER_REQUEST tables"
                .to_string(),
        ));
    }

    for (i, vix) in vixs.iter().enumerate() {
        crate::validation::validate_vector_index(
            vix,
            defs,
            &format!("vectorIndexes.{}.member", i + 1),
        )?;
    }

    // Duplicate index names within the vector list.
    let mut seen = std::collections::HashSet::new();
    for vix in vixs {
        if !seen.insert(&vix.index_name) {
            return Err(ve(format!(
                "One or more parameter values were invalid: Duplicate index name: {}",
                vix.index_name
            )));
        }
    }

    // Count limit (five per table).
    if vixs.len() > 5 {
        return Err(ve(
            "One or more parameter values were invalid: VectorIndex count exceeds the \
             per-table limit of 5"
                .to_string(),
        ));
    }

    // Two indexes on one vector attribute must agree on dimensions.
    let mut dims_by_attr: std::collections::HashMap<&str, u32> = std::collections::HashMap::new();
    for vix in vixs {
        let attr = vix.vector_attribute.attribute_name.as_str();
        match dims_by_attr.get(attr) {
            Some(&prev) if prev != vix.dimensions => {
                return Err(ve(format!(
                    "One or more parameter values were invalid: Conflicting attribute \
                     definition for '{attr}'. All VectorIndexes on the same vector attribute \
                     must use the same dimensions."
                )));
            }
            Some(_) => {}
            None => {
                dims_by_attr.insert(attr, vix.dimensions);
            }
        }
    }

    // The vector attribute must not itself be declared in AttributeDefinitions:
    // it is not a key attribute, and declaring it is rejected outright.
    // Captured from real DynamoDB (eu-west-2 and us-east-1, 2026-08-12).
    for vix in vixs {
        let attr = vix.vector_attribute.attribute_name.as_str();
        if defs.iter().any(|d| d.attribute_name == attr) {
            return Err(ve(format!(
                "One or more parameter values were invalid: Conflicting attribute definition \
                 for '{attr}'. An attribute cannot be defined in AttributeDefinitions when \
                 used as a VectorAttribute."
            )));
        }
    }

    Ok(())
}

/// Duplicate index names across the LSI, GSI, and vector index families.
/// LSIs, GSIs, and vector indexes share one name namespace; the classic
/// duplicate message covers a vector index colliding with either secondary
/// index family too (captured from real DynamoDB, eu-west-2 and us-east-1,
/// 2026-08-12). Duplicates within a single family are reported by that
/// family's own list validation, which runs first.
fn check_cross_index_duplicates(
    lsis: &Option<Vec<LocalSecondaryIndex>>,
    gsis: &Option<Vec<GlobalSecondaryIndex>>,
    vixs: &Option<Vec<VectorIndex>>,
) -> std::result::Result<(), String> {
    let mut all_names = std::collections::HashSet::new();
    if let Some(lsis) = lsis {
        for lsi in lsis {
            all_names.insert(&lsi.index_name);
        }
    }
    if let Some(gsis) = gsis {
        for gsi in gsis {
            if !all_names.insert(&gsi.index_name) {
                return Err(format!(
                    "One or more parameter values were invalid: Duplicate index name: {}",
                    gsi.index_name
                ));
            }
        }
    }
    if let Some(vixs) = vixs {
        for vix in vixs {
            if !all_names.insert(&vix.index_name) {
                return Err(format!(
                    "One or more parameter values were invalid: Duplicate index name: {}",
                    vix.index_name
                ));
            }
        }
    }
    Ok(())
}

// ---- Raw JSON validation (for deserialization path) ----

fn validate_raw_and_build(raw: RawRequest) -> std::result::Result<CreateTableRequest, String> {
    // Missing TableName is a different error format from invalid TableName
    if raw.table_name.is_none() {
        return Err(
            "The parameter 'TableName' is required but was not present in the request".to_string(),
        );
    }

    // Use the shared constraint error collector for table name validation.
    // This produces the correct multi-field constraint format for empty,
    // too-short, too-long, or invalid-pattern table names.
    let name_errors = crate::validation::table_name_constraint_errors(
        raw.table_name.as_deref(),
        crate::validation::TableNameContext::CreateTable,
    );
    if !name_errors.is_empty() {
        let msg = format!(
            "{} validation error{} detected: {}",
            name_errors.len(),
            if name_errors.len() > 1 { "s" } else { "" },
            name_errors.join("; ")
        );
        return Err(msg);
    }
    let table_name = raw.table_name.unwrap();

    let mut errors = Vec::new();

    if let Some(ref bm) = raw.billing_mode
        && bm != "PROVISIONED"
        && bm != "PAY_PER_REQUEST"
    {
        errors.push(format!(
            "Value '{}' at 'billingMode' failed to satisfy constraint: \
                 Member must satisfy enum value set: [PROVISIONED, PAY_PER_REQUEST]",
            bm
        ));
    }

    collect_pt_errors(&raw.provisioned_throughput, &mut errors);
    collect_ks_errors(&raw.key_schema, &mut errors);
    collect_ad_errors(&raw.attribute_definitions, &mut errors);
    // The GSIs' member errors come before the LSIs': with a too-short name on
    // both, DynamoDB lists the GSI's first, whichever order the request gives
    // them in (captured eu-west-2 and us-east-1, 2026-10-09).
    collect_gsi_errors(&raw.global_secondary_indexes, &mut errors);
    collect_lsi_errors(&raw.local_secondary_indexes, &mut errors);
    collect_vix_errors(&raw.vector_indexes, &mut errors);

    // DynamoDB caps multi-field constraint errors at 10
    errors.truncate(10);

    if !errors.is_empty() {
        let prefix = format!(
            "{} validation error{} detected: ",
            errors.len(),
            if errors.len() == 1 { "" } else { "s" }
        );
        return Err(format!("{}{}", prefix, errors.join("; ")));
    }

    // BillingMode + ProvisionedThroughput consistency (HTTP path only)
    let billing_mode_str = raw.billing_mode.as_deref().unwrap_or("PROVISIONED");
    if billing_mode_str == "PAY_PER_REQUEST" && raw.provisioned_throughput.is_some() {
        return Err(
            "One or more parameter values were invalid: Neither ReadCapacityUnits nor \
             WriteCapacityUnits can be specified when BillingMode is PAY_PER_REQUEST"
                .to_string(),
        );
    }

    // ProvisionedThroughput out-of-bounds (after multi-field but before struct checks)
    if let Some(ref pt) = raw.provisioned_throughput
        && let Some(obj) = pt.as_object()
    {
        let rcu = obj
            .get("ReadCapacityUnits")
            .and_then(|v| v.as_i64())
            .unwrap_or(0);
        let wcu = obj
            .get("WriteCapacityUnits")
            .and_then(|v| v.as_i64())
            .unwrap_or(0);
        const MAX_THROUGHPUT: i64 = 1_000_000_000_000;
        if rcu > MAX_THROUGHPUT {
            return Err(format!(
                "Given value {} for ReadCapacityUnits is out of bounds",
                rcu
            ));
        }
        if wcu > MAX_THROUGHPUT {
            return Err(format!(
                "Given value {} for WriteCapacityUnits is out of bounds",
                wcu
            ));
        }
    }

    // Missing ProvisionedThroughput when BillingMode is explicitly PROVISIONED.
    if raw.billing_mode.as_deref() == Some("PROVISIONED") && raw.provisioned_throughput.is_none() {
        return Err(
            "One or more parameter values were invalid: ReadCapacityUnits and \
             WriteCapacityUnits must both be specified when BillingMode is PROVISIONED"
                .to_string(),
        );
    }

    // Parse typed structures
    let key_schema: Vec<KeySchemaElement> = raw
        .key_schema
        .as_ref()
        .map(|v| serde_json::from_value(v.clone()))
        .transpose()
        .map_err(|e| e.to_string())?
        .unwrap_or_default();
    let attribute_definitions: Vec<AttributeDefinition> = raw
        .attribute_definitions
        .as_ref()
        .map(|v| serde_json::from_value(v.clone()))
        .transpose()
        .map_err(|e| e.to_string())?
        .unwrap_or_default();
    let provisioned_throughput: Option<ProvisionedThroughput> = raw
        .provisioned_throughput
        .as_ref()
        .map(|v| serde_json::from_value(v.clone()))
        .transpose()
        .map_err(|e| e.to_string())?;
    let global_secondary_indexes: Option<Vec<GlobalSecondaryIndex>> = raw
        .global_secondary_indexes
        .as_ref()
        .map(|v| serde_json::from_value(v.clone()))
        .transpose()
        .map_err(|e| e.to_string())?;
    let local_secondary_indexes: Option<Vec<LocalSecondaryIndex>> = raw
        .local_secondary_indexes
        .as_ref()
        .map(|v| serde_json::from_value(v.clone()))
        .transpose()
        .map_err(|e| e.to_string())?;
    let vector_indexes: Option<Vec<VectorIndex>> = raw
        .vector_indexes
        .as_ref()
        .map(|v| serde_json::from_value(normalise_vix_dimensions(v)))
        .transpose()
        .map_err(|e| e.to_string())?;

    Ok(CreateTableRequest {
        table_name,
        key_schema,
        attribute_definitions,
        global_secondary_indexes,
        local_secondary_indexes,
        vector_indexes,
        billing_mode: raw.billing_mode,
        provisioned_throughput,
        stream_specification: raw.stream_specification,
        sse_specification: raw.sse_specification,
        table_class: raw.table_class,
        tags: raw.tags,
        deletion_protection_enabled: raw.deletion_protection_enabled,
        on_demand_throughput: raw.on_demand_throughput,
    })
}

// ---- Multi-field constraint error collectors ----

fn collect_pt_errors(pt_val: &Option<serde_json::Value>, errors: &mut Vec<String>) {
    if let Some(v) = pt_val
        && let Some(obj) = v.as_object()
    {
        let wcu = obj.get("WriteCapacityUnits");
        let rcu = obj.get("ReadCapacityUnits");
        if wcu.is_none() || wcu == Some(&serde_json::Value::Null) {
            errors.push("Value null at 'provisionedThroughput.writeCapacityUnits' failed to satisfy constraint: Member must not be null".to_string());
        } else if let Some(w) = wcu.and_then(|v| v.as_i64())
            && w < 1
        {
            errors.push(format!("Value '{}' at 'provisionedThroughput.writeCapacityUnits' failed to satisfy constraint: Member must have value greater than or equal to 1", w));
        }
        if rcu.is_none() || rcu == Some(&serde_json::Value::Null) {
            errors.push("Value null at 'provisionedThroughput.readCapacityUnits' failed to satisfy constraint: Member must not be null".to_string());
        } else if let Some(r) = rcu.and_then(|v| v.as_i64())
            && r < 1
        {
            errors.push(format!("Value '{}' at 'provisionedThroughput.readCapacityUnits' failed to satisfy constraint: Member must have value greater than or equal to 1", r));
        }
    }
}

fn collect_ks_errors(ks_val: &Option<serde_json::Value>, errors: &mut Vec<String>) {
    match ks_val {
        None => {
            errors.push(
                "Value null at 'keySchema' failed to satisfy constraint: Member must not be null"
                    .to_string(),
            );
        }
        Some(v) => {
            if let Some(arr) = v.as_array() {
                if arr.is_empty() {
                    errors.push("Value '[]' at 'keySchema' failed to satisfy constraint: Member must have length greater than or equal to 1".to_string());
                } else if arr.len() > 2 {
                    let dump = render_key_schema_java_toString(arr);
                    errors.push(format!("Value '{}' at 'keySchema' failed to satisfy constraint: Member must have length less than or equal to 2", dump));
                }
                for (i, elem) in arr.iter().enumerate().take(10) {
                    collect_ks_elem_errors(elem, i + 1, errors);
                }
            }
        }
    }
}

fn collect_ks_elem_errors(elem: &serde_json::Value, idx: usize, errors: &mut Vec<String>) {
    if let Some(obj) = elem.as_object() {
        if !obj.contains_key("AttributeName")
            || obj.get("AttributeName") == Some(&serde_json::Value::Null)
        {
            errors.push(format!("Value null at 'keySchema.{}.member.attributeName' failed to satisfy constraint: Member must not be null", idx));
        }
        let kt = obj.get("KeyType");
        if kt.is_none() || kt == Some(&serde_json::Value::Null) {
            errors.push(format!("Value null at 'keySchema.{}.member.keyType' failed to satisfy constraint: Member must not be null", idx));
        } else if let Some(s) = kt.and_then(|v| v.as_str())
            && s != "HASH"
            && s != "RANGE"
        {
            errors.push(format!("Value '{}' at 'keySchema.{}.member.keyType' failed to satisfy constraint: Member must satisfy enum value set: [HASH, RANGE]", s, idx));
        }
    }
}

/// Render a KeySchema array the way AWS does in validation messages: Java's
/// default toString() shape over the SDK's KeySchemaElement model, e.g.
/// `[KeySchemaElement(attributeName=pk, keyType=HASH), KeySchemaElement(attributeName=sk, keyType=RANGE)]`.
/// Missing attribute fields render as the empty string, matching what AWS
/// emits when the upstream object hasn't been populated.
#[allow(non_snake_case)]
fn render_key_schema_java_toString(arr: &[serde_json::Value]) -> String {
    let parts: Vec<String> = arr
        .iter()
        .map(|elem| {
            let an = elem
                .get("AttributeName")
                .and_then(|v| v.as_str())
                .unwrap_or("");
            let kt = elem.get("KeyType").and_then(|v| v.as_str()).unwrap_or("");
            format!("KeySchemaElement(attributeName={an}, keyType={kt})")
        })
        .collect();
    format!("[{}]", parts.join(", "))
}

fn collect_ad_errors(ad_val: &Option<serde_json::Value>, errors: &mut Vec<String>) {
    match ad_val {
        None => {
            errors.push("Value null at 'attributeDefinitions' failed to satisfy constraint: Member must not be null".to_string());
        }
        Some(v) => {
            if let Some(arr) = v.as_array() {
                for (i, elem) in arr.iter().enumerate() {
                    if let Some(obj) = elem.as_object() {
                        if !obj.contains_key("AttributeName")
                            || obj.get("AttributeName") == Some(&serde_json::Value::Null)
                        {
                            errors.push(format!("Value null at 'attributeDefinitions.{}.member.attributeName' failed to satisfy constraint: Member must not be null", i + 1));
                        }
                        let at = obj.get("AttributeType");
                        if at.is_none() || at == Some(&serde_json::Value::Null) {
                            errors.push(format!("Value null at 'attributeDefinitions.{}.member.attributeType' failed to satisfy constraint: Member must not be null", i + 1));
                        } else if let Some(s) = at.and_then(|v| v.as_str())
                            && s != "S"
                            && s != "N"
                            && s != "B"
                        {
                            errors.push(format!("Value '{}' at 'attributeDefinitions.{}.member.attributeType' failed to satisfy constraint: Member must satisfy enum value set: [B, N, S]", s, i + 1));
                        }
                    }
                }
            }
        }
    }
}

fn collect_lsi_errors(lsi_val: &Option<serde_json::Value>, errors: &mut Vec<String>) {
    if let Some(v) = lsi_val
        && let Some(arr) = v.as_array()
    {
        for (i, elem) in arr.iter().enumerate().take(10) {
            if let Some(obj) = elem.as_object() {
                // Order: indexName, keySchema, projection
                if !obj.contains_key("IndexName")
                    || obj.get("IndexName") == Some(&serde_json::Value::Null)
                {
                    errors.push(format!("Value null at 'localSecondaryIndexes.{}.member.indexName' failed to satisfy constraint: Member must not be null", i + 1));
                } else if let Some(name) = obj.get("IndexName").and_then(|v| v.as_str()) {
                    collect_idx_name_errors(name, "localSecondaryIndexes", i + 1, errors);
                }
                if !obj.contains_key("KeySchema")
                    || obj.get("KeySchema") == Some(&serde_json::Value::Null)
                {
                    errors.push(format!("Value null at 'localSecondaryIndexes.{}.member.keySchema' failed to satisfy constraint: Member must not be null", i + 1));
                } else if let Some(ks) = obj.get("KeySchema").and_then(|v| v.as_array())
                    && ks.is_empty()
                {
                    errors.push(format!("Value '[]' at 'localSecondaryIndexes.{}.member.keySchema' failed to satisfy constraint: Member must have length greater than or equal to 1", i + 1));
                } else if let Some(ks) = obj.get("KeySchema").and_then(|v| v.as_array())
                    && ks.len() > 2
                {
                    // An LSI key is the table's partition key and one sort key
                    // attribute. AWS names the path `keySchema` here, not the
                    // index member's (captured eu-west-2, 2026-10-09).
                    errors.push(format!("Value '{}' at 'keySchema' failed to satisfy constraint: Member must have length less than or equal to 2", render_key_schema_java_toString(ks)));
                }
                if !obj.contains_key("Projection")
                    || obj.get("Projection") == Some(&serde_json::Value::Null)
                {
                    errors.push(format!("Value null at 'localSecondaryIndexes.{}.member.projection' failed to satisfy constraint: Member must not be null", i + 1));
                } else if let Some(p) = obj.get("Projection").and_then(|v| v.as_object()) {
                    collect_proj_errors(
                        p,
                        &format!("localSecondaryIndexes.{}.member", i + 1),
                        errors,
                    );
                }
            }
        }
    }
}

fn collect_gsi_errors(gsi_val: &Option<serde_json::Value>, errors: &mut Vec<String>) {
    if let Some(v) = gsi_val
        && let Some(arr) = v.as_array()
    {
        for (i, elem) in arr.iter().enumerate().take(10) {
            if let Some(obj) = elem.as_object() {
                // Order for GSI: keySchema, projection, indexName
                if !obj.contains_key("KeySchema")
                    || obj.get("KeySchema") == Some(&serde_json::Value::Null)
                {
                    errors.push(format!("Value null at 'globalSecondaryIndexes.{}.member.keySchema' failed to satisfy constraint: Member must not be null", i + 1));
                } else if let Some(ks) = obj.get("KeySchema").and_then(|v| v.as_array())
                    && ks.is_empty()
                {
                    errors.push(format!("Value '[]' at 'globalSecondaryIndexes.{}.member.keySchema' failed to satisfy constraint: Member must have length greater than or equal to 1", i + 1));
                } else if let Some(ks) = obj.get("KeySchema").and_then(|v| v.as_array())
                    && ks.len() > 8
                {
                    // Four partition key and four sort key attributes at most.
                    // AWS names the path `keySchema` here, as it does for an
                    // LSI (captured eu-west-2 and us-east-1, 2026-10-09).
                    errors.push(format!("Value '{}' at 'keySchema' failed to satisfy constraint: Member must have length less than or equal to 8", render_key_schema_java_toString(ks)));
                }
                if !obj.contains_key("Projection")
                    || obj.get("Projection") == Some(&serde_json::Value::Null)
                {
                    errors.push(format!("Value null at 'globalSecondaryIndexes.{}.member.projection' failed to satisfy constraint: Member must not be null", i + 1));
                } else if let Some(p) = obj.get("Projection").and_then(|v| v.as_object()) {
                    collect_proj_errors(
                        p,
                        &format!("globalSecondaryIndexes.{}.member", i + 1),
                        errors,
                    );
                }
                if !obj.contains_key("IndexName")
                    || obj.get("IndexName") == Some(&serde_json::Value::Null)
                {
                    errors.push(format!("Value null at 'globalSecondaryIndexes.{}.member.indexName' failed to satisfy constraint: Member must not be null", i + 1));
                } else if let Some(name) = obj.get("IndexName").and_then(|v| v.as_str()) {
                    collect_idx_name_errors(name, "globalSecondaryIndexes", i + 1, errors);
                }
                // GSI ProvisionedThroughput
                if let Some(pt) = obj.get("ProvisionedThroughput").and_then(|v| v.as_object()) {
                    let wcu = pt.get("WriteCapacityUnits");
                    let rcu = pt.get("ReadCapacityUnits");
                    if let Some(w) = wcu.and_then(|v| v.as_i64()) {
                        if w < 1 {
                            errors.push(format!("Value '{}' at 'globalSecondaryIndexes.{}.member.provisionedThroughput.writeCapacityUnits' failed to satisfy constraint: Member must have value greater than or equal to 1", w, i + 1));
                        }
                    } else if wcu.is_none() || wcu == Some(&serde_json::Value::Null) {
                        errors.push(format!("Value null at 'globalSecondaryIndexes.{}.member.provisionedThroughput.writeCapacityUnits' failed to satisfy constraint: Member must not be null", i + 1));
                    }
                    if let Some(r) = rcu.and_then(|v| v.as_i64()) {
                        if r < 1 {
                            errors.push(format!("Value '{}' at 'globalSecondaryIndexes.{}.member.provisionedThroughput.readCapacityUnits' failed to satisfy constraint: Member must have value greater than or equal to 1", r, i + 1));
                        }
                    } else if rcu.is_none() || rcu == Some(&serde_json::Value::Null) {
                        errors.push(format!("Value null at 'globalSecondaryIndexes.{}.member.provisionedThroughput.readCapacityUnits' failed to satisfy constraint: Member must not be null", i + 1));
                    }
                }
            }
        }
    }
}

/// Run the `VectorIndexes` request-model constraints and format the envelope
/// DynamoDB answers with, for a caller that does not arrive through the raw
/// request's `Deserialize`.
///
/// The MCP surface builds its request field by field, so without this it skips
/// every constraint collected here. Feed it the canonical PascalCase form: the
/// collectors read the wire spelling, and MCP names its fields in snake_case,
/// so re-serialising the typed value first is what makes the two agree.
#[cfg(feature = "mcp-server")]
pub(crate) fn vector_indexes_request_model_error(
    vix_val: &Option<serde_json::Value>,
) -> Option<String> {
    let mut errors = Vec::new();
    collect_vix_errors(vix_val, &mut errors);
    errors.truncate(10);
    if errors.is_empty() {
        return None;
    }
    Some(format!(
        "{} validation error{} detected: {}",
        errors.len(),
        if errors.len() == 1 { "" } else { "s" },
        errors.join("; ")
    ))
}

/// Request-model constraint errors for `VectorIndexes`, in API-model member
/// order: indexName, vectorAttribute, searchSchema, projection, dimensions,
/// distanceFunction. The `Dimensions` lower-bound form is captured from real
/// DynamoDB (eu-west-2, 2026-08-11); the rest follow the formulaic constraint
/// pattern the neighbouring collectors pin.
fn collect_vix_errors(vix_val: &Option<serde_json::Value>, errors: &mut Vec<String>) {
    let arr = match vix_val.as_ref().and_then(|v| v.as_array()) {
        Some(arr) => arr,
        None => return,
    };
    for (i, elem) in arr.iter().enumerate().take(10) {
        let obj = match elem.as_object() {
            Some(obj) => obj,
            None => continue,
        };
        collect_vix_obj_errors(obj, &format!("vectorIndexes.{}.member", i + 1), errors);
    }
}

/// Request-model constraint errors for one raw vector index object, shared
/// with the `UpdateTable` create-action collector: `path` is the object's
/// member path (`vectorIndexes.N.member` here,
/// `vectorIndexUpdates.N.member.create` there, the latter captured from real
/// DynamoDB, eu-west-2 and us-east-1, 2026-08-12).
pub(crate) fn collect_vix_obj_errors(
    obj: &serde_json::Map<String, serde_json::Value>,
    path: &str,
    errors: &mut Vec<String>,
) {
    if !obj.contains_key("IndexName") || obj.get("IndexName") == Some(&serde_json::Value::Null) {
        errors.push(format!(
            "Value null at '{path}.indexName' failed to satisfy constraint: Member must not be null"
        ));
    } else if let Some(name) = obj.get("IndexName").and_then(|v| v.as_str()) {
        collect_idx_name_errors_at(name, path, errors);
    }
    if !obj.contains_key("VectorAttribute")
        || obj.get("VectorAttribute") == Some(&serde_json::Value::Null)
    {
        errors.push(format!(
            "Value null at '{path}.vectorAttribute' failed to satisfy constraint: Member must not be null"
        ));
    } else if let Some(va) = obj.get("VectorAttribute").and_then(|v| v.as_object()) {
        if !va.contains_key("AttributeName")
            || va.get("AttributeName") == Some(&serde_json::Value::Null)
        {
            errors.push(format!(
                "Value null at '{path}.vectorAttribute.attributeName' failed to satisfy constraint: Member must not be null"
            ));
        } else if let Some(name) = va.get("AttributeName").and_then(|v| v.as_str()) {
            // The API model bounds this at one character, which the AWS CLI
            // enforces before the request leaves the client. Without it an
            // empty name was accepted, the index reported ACTIVE, and no item
            // could ever derive a row, so the index stayed empty for good.
            if name.is_empty() {
                errors.push(format!(
                    "Value '' at '{path}.vectorAttribute.attributeName' failed to satisfy constraint: Member must have length greater than or equal to 1"
                ));
            }
        }
    }
    if let Some(schema) = obj.get("SearchSchema").and_then(|v| v.as_array()) {
        for (j, se) in schema.iter().enumerate().take(10) {
            let seo = match se.as_object() {
                Some(seo) => seo,
                None => continue,
            };
            let sidx = j + 1;
            if !seo.contains_key("AttributeName")
                || seo.get("AttributeName") == Some(&serde_json::Value::Null)
            {
                errors.push(format!(
                    "Value null at '{path}.searchSchema.{sidx}.member.attributeName' failed to satisfy constraint: Member must not be null"
                ));
            }
            let et = seo.get("SearchSchemaElementType");
            if et.is_none() || et == Some(&serde_json::Value::Null) {
                errors.push(format!(
                    "Value null at '{path}.searchSchema.{sidx}.member.searchSchemaElementType' failed to satisfy constraint: Member must not be null"
                ));
            } else if let Some(s) = et.and_then(|v| v.as_str())
                && s != "HASH"
                && s != "INLINE_FILTER"
            {
                errors.push(format!(
                        "Value '{s}' at '{path}.searchSchema.{sidx}.member.searchSchemaElementType' failed to satisfy constraint: Member must satisfy enum value set: [HASH, INLINE_FILTER]"
                    ));
            }
        }
    }
    if !obj.contains_key("Projection") || obj.get("Projection") == Some(&serde_json::Value::Null) {
        errors.push(format!(
            "Value null at '{path}.projection' failed to satisfy constraint: Member must not be null"
        ));
    } else if let Some(p) = obj.get("Projection").and_then(|v| v.as_object()) {
        collect_proj_errors(p, path, errors);
    }
    let dims = obj.get("Dimensions");
    if dims.is_none() || dims == Some(&serde_json::Value::Null) {
        errors.push(format!(
            "Value null at '{path}.dimensions' failed to satisfy constraint: Member must not be null"
        ));
    } else if let Some(d) = dims.and_then(truncated_dimensions) {
        // Fractional values are truncated before validation (captured
        // from real DynamoDB, eu-west-2 and us-east-1, 2026-08-12), so
        // the lower bound applies to the truncated value; the upper
        // bound is enforced later by the typed validator with the bare
        // operation-layer message.
        if d < 1 {
            errors.push(format!(
                "Value '{d}' at '{path}.dimensions' failed to satisfy constraint: Member must have value greater than or equal to 1"
            ));
        }
    }
    let df = obj.get("DistanceFunction");
    if df.is_none() || df == Some(&serde_json::Value::Null) {
        errors.push(format!(
            "Value null at '{path}.distanceFunction' failed to satisfy constraint: Member must not be null"
        ));
    } else if let Some(s) = df.and_then(|v| v.as_str())
        && s != "COSINE"
        && s != "DOT_PRODUCT"
        && s != "EUCLIDEAN"
    {
        errors.push(format!(
                "Value '{s}' at '{path}.distanceFunction' failed to satisfy constraint: Member must satisfy enum value set: [COSINE, DOT_PRODUCT, EUCLIDEAN]"
            ));
    }
}

/// Effective integer value of a raw `Dimensions` number, truncating any
/// fractional part (captured from real DynamoDB, eu-west-2 and us-east-1,
/// 2026-08-12: Dimensions 3.5 is accepted and DescribeTable reports 3).
/// Values beyond the i64 range saturate; they are far above the 4096 ceiling
/// either way. Returns `None` for non-numeric values.
fn truncated_dimensions(v: &serde_json::Value) -> Option<i64> {
    if let Some(d) = v.as_i64() {
        return Some(d);
    }
    if v.as_u64().is_some() {
        // Only reachable for integers above the i64 range.
        return Some(i64::MAX);
    }
    // `as` saturates at the i64 bounds for out-of-range floats.
    v.as_f64().map(|f| f.trunc() as i64)
}

/// Normalise raw `Dimensions` values so the typed `u32` parse cannot fail on
/// any numeric input. Fractional values are truncated (see
/// [`truncated_dimensions`]); integers above the u32 range clamp to
/// `u32::MAX`, which is over the 4096 ceiling, so the typed validator then
/// rejects them with the captured over-range message rather than letting a
/// raw serde error surface.
pub(crate) fn normalise_vix_dimensions(vix_val: &serde_json::Value) -> serde_json::Value {
    let mut val = vix_val.clone();
    if let Some(arr) = val.as_array_mut() {
        for elem in arr {
            if let Some(obj) = elem.as_object_mut() {
                normalise_vix_dimensions_obj(obj);
            }
        }
    }
    val
}

/// Normalise one raw vector index object's `Dimensions` value in place, shared
/// with the `UpdateTable` create-action path so both parse routes apply the
/// same truncation and clamping before the typed `u32` parse.
pub(crate) fn normalise_vix_dimensions_obj(obj: &mut serde_json::Map<String, serde_json::Value>) {
    // The MCP surface names its fields in snake_case, which `VectorIndex`
    // accepts by alias, so the normalisation has to recognise both spellings or
    // an agent would be refused a value a wire client gets truncated.
    let key = if obj.contains_key("Dimensions") {
        "Dimensions"
    } else {
        "dimensions"
    };
    let Some(dims) = obj.get_mut(key) else {
        return;
    };
    if let Some(d) = truncated_dimensions(dims) {
        *dims = serde_json::Value::from(d.clamp(0, i64::from(u32::MAX)) as u32);
    }
}

fn collect_idx_name_errors(name: &str, prefix: &str, idx: usize, errors: &mut Vec<String>) {
    collect_idx_name_errors_at(name, &format!("{prefix}.{idx}.member"), errors);
}

/// As [`collect_idx_name_errors`], but taking the full member path directly,
/// for callers whose path does not follow the `<list>.<idx>.member` shape.
fn collect_idx_name_errors_at(name: &str, path: &str, errors: &mut Vec<String>) {
    if !name
        .chars()
        .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-' || c == '.')
    {
        errors.push(format!("Value '{}' at '{}.indexName' failed to satisfy constraint: Member must satisfy regular expression pattern: [a-zA-Z0-9_.-]+", name, path));
    }
    if name.len() < 3 {
        errors.push(format!("Value '{}' at '{}.indexName' failed to satisfy constraint: Member must have length greater than or equal to 3", name, path));
    }
    if name.len() > 255 {
        errors.push(format!("Value '{}' at '{}.indexName' failed to satisfy constraint: Member must have length less than or equal to 255", name, path));
    }
}

/// `prefix` is the owning index object's full member path (for example
/// `globalSecondaryIndexes.1.member`), so callers with a deeper shape can
/// pass their own.
fn collect_proj_errors(
    proj: &serde_json::Map<String, serde_json::Value>,
    prefix: &str,
    errors: &mut Vec<String>,
) {
    if let Some(pt) = proj.get("ProjectionType")
        && let Some(s) = pt.as_str()
        && s != "ALL"
        && s != "KEYS_ONLY"
        && s != "INCLUDE"
    {
        errors.push(format!("Value '{}' at '{}.projection.projectionType' failed to satisfy constraint: Member must satisfy enum value set: [ALL, INCLUDE, KEYS_ONLY]", s, prefix));
    }
    if let Some(nka) = proj.get("NonKeyAttributes")
        && let Some(arr) = nka.as_array()
        && arr.is_empty()
    {
        errors.push(format!("Value '[]' at '{}.projection.nonKeyAttributes' failed to satisfy constraint: Member must have length greater than or equal to 1", prefix));
    }
}

// ---- Structural validation helpers ----

fn validate_key_schema_structure(ks: &[KeySchemaElement]) -> std::result::Result<(), String> {
    if ks.is_empty() {
        return Err("1 validation error detected: Value null at 'keySchema' failed to satisfy constraint: Member must have length less than or equal to 2".to_string());
    }
    if ks[0].key_type != KeyType::HASH {
        return Err(
            "Invalid KeySchema: The first KeySchemaElement is not a HASH key type".to_string(),
        );
    }
    if ks.len() == 2 && ks[0].attribute_name == ks[1].attribute_name {
        return Err(
            "Both the Hash Key and the Range Key element in the KeySchema have the same name"
                .to_string(),
        );
    }
    if ks.len() == 2 && ks[1].key_type != KeyType::RANGE {
        return Err(
            "Invalid KeySchema: The second KeySchemaElement is not a RANGE key type".to_string(),
        );
    }
    Ok(())
}

fn validate_key_attrs_in_defs(
    ks: &[KeySchemaElement],
    defs: &[AttributeDefinition],
) -> std::result::Result<(), String> {
    // Collect missing key attribute names
    let missing: Vec<&str> = ks
        .iter()
        .filter(|k| !defs.iter().any(|d| d.attribute_name == k.attribute_name))
        .map(|k| k.attribute_name.as_str())
        .collect();

    if missing.is_empty() {
        // Even if no keys are missing, check for structural issues (dup names/types)
        // which DynamoDB reports as generic "no definition" when defs exist
        let has_dup_names = ks.len() == 2 && ks[0].attribute_name == ks[1].attribute_name;
        if has_dup_names {
            return Err(
                "Invalid KeySchema: Some index key attribute have no definition".to_string(),
            );
        }
        return Ok(());
    }

    // Use generic message when:
    // - defs is empty (fewer defs than unique key attrs)
    // - key schema has 2 elements (regardless of structural validity)
    // - key schema has structural issues (dup names, dup types)
    // Use detailed message only when defs is non-empty AND key schema has 1 element
    let has_dup_names = ks.len() == 2 && ks[0].attribute_name == ks[1].attribute_name;
    let has_dup_types = ks.len() == 2 && ks[0].key_type == ks[1].key_type;
    let use_generic = defs.is_empty() || ks.len() >= 2 || has_dup_names || has_dup_types;

    if use_generic {
        return Err("Invalid KeySchema: Some index key attribute have no definition".to_string());
    }

    // Detailed message for single-key schema with non-empty defs
    let key_names: Vec<&str> = missing.to_vec();
    let def_names: Vec<&str> = defs.iter().map(|d| d.attribute_name.as_str()).collect();
    Err(format!(
        "One or more parameter values were invalid: Some index key attributes are not defined in \
         AttributeDefinitions. Keys: [{}], AttributeDefinitions: [{}]",
        key_names.join(", "),
        def_names.join(", ")
    ))
}

fn validate_attr_def_count(
    ks: &[KeySchemaElement],
    defs: &[AttributeDefinition],
    lsis: &Option<Vec<LocalSecondaryIndex>>,
    gsis: &Option<Vec<GlobalSecondaryIndex>>,
    vixs: &Option<Vec<VectorIndex>>,
) -> std::result::Result<(), String> {
    let mut all_key_attrs = std::collections::HashSet::new();
    for k in ks {
        all_key_attrs.insert(k.attribute_name.as_str());
    }
    if let Some(lsis) = lsis {
        for lsi in lsis {
            for k in &lsi.key_schema {
                all_key_attrs.insert(k.attribute_name.as_str());
            }
        }
    }
    if let Some(gsis) = gsis {
        for gsi in gsis {
            for k in &gsi.key_schema {
                all_key_attrs.insert(k.attribute_name.as_str());
            }
        }
    }
    // SearchSchema attributes must be declared in AttributeDefinitions, so
    // they join the used set; the vector attribute itself is not a key
    // attribute and never counts.
    if let Some(vixs) = vixs {
        for vix in vixs {
            if let Some(ref schema) = vix.search_schema {
                for elem in schema {
                    all_key_attrs.insert(elem.attribute_name.as_str());
                }
            }
        }
    }
    if defs.len() != all_key_attrs.len() {
        let has_index = lsis.as_ref().is_some_and(|l| !l.is_empty())
            || gsis.as_ref().is_some_and(|g| !g.is_empty());
        if !has_index {
            return Err("One or more parameter values were invalid: Number of attributes in KeySchema does not exactly match number of attributes defined in AttributeDefinitions".to_string());
        }
        // With an index, both sets are listed, each in the order DynamoDB's own
        // hash tables give them. Captured eu-west-2 and us-east-1, 2026-10-09.
        let defined: Vec<&str> = defs.iter().map(|d| d.attribute_name.as_str()).collect();
        let mut used: Vec<&str> = ks.iter().map(|k| k.attribute_name.as_str()).collect();
        for k in lsis.iter().flatten().flat_map(|l| &l.key_schema) {
            used.push(k.attribute_name.as_str());
        }
        for k in gsis.iter().flatten().flat_map(|g| &g.key_schema) {
            used.push(k.attribute_name.as_str());
        }
        for elem in vixs
            .iter()
            .flatten()
            .filter_map(|v| v.search_schema.as_ref())
            .flatten()
        {
            used.push(elem.attribute_name.as_str());
        }
        return Err(format!(
            "One or more parameter values were invalid: Some AttributeDefinitions are not used. \
             AttributeDefinitions: [{}], keys used: [{}]",
            crate::validation::java_hash_order(&defined, true).join(", "),
            crate::validation::java_hash_order(&used, false).join(", ")
        ));
    }
    Ok(())
}

fn validate_lsi_list(
    lsis: &[LocalSecondaryIndex],
    ks: &[KeySchemaElement],
    defs: &[AttributeDefinition],
) -> std::result::Result<(), String> {
    // Empty check is done earlier in validate_typed_request

    if !ks.iter().any(|k| k.key_type == KeyType::RANGE) {
        return Err("One or more parameter values were invalid: Table KeySchema does not have a range key, which is required when specifying a LocalSecondaryIndex".to_string());
    }

    // The first index naming an attribute the request does not define.
    for lsi in lsis {
        if let Some(message) = crate::validation::undefined_index_key_error(&lsi.key_schema, defs) {
            return Err(message);
        }
    }

    // Structural validation for each LSI
    for lsi in lsis {
        validate_lsi_structure(lsi, ks)?;
    }

    // Duplicate index names
    let mut seen = std::collections::HashSet::new();
    for lsi in lsis {
        if !seen.insert(&lsi.index_name) {
            return Err(format!(
                "One or more parameter values were invalid: Duplicate index name: {}",
                lsi.index_name
            ));
        }
    }

    // Count limit
    if lsis.len() > 5 {
        return Err("One or more parameter values were invalid: Number of LocalSecondaryIndexes exceeds per-table limit of 5".to_string());
    }

    Ok(())
}

fn validate_gsi_list(
    gsis: &[GlobalSecondaryIndex],
    defs: &[AttributeDefinition],
    bm: &str,
) -> std::result::Result<(), String> {
    // Empty check is done earlier in validate_typed_request

    // The first index naming an attribute the request does not define.
    for gsi in gsis {
        if let Some(message) = crate::validation::undefined_index_key_error(&gsi.key_schema, defs) {
            return Err(message);
        }
    }

    // Structural validation for each GSI
    for gsi in gsis {
        validate_gsi_structure(gsi)?;
    }

    // Duplicate index names
    let mut seen = std::collections::HashSet::new();
    for gsi in gsis {
        if !seen.insert(&gsi.index_name) {
            return Err(format!(
                "One or more parameter values were invalid: Duplicate index name: {}",
                gsi.index_name
            ));
        }
    }

    // Count limit
    if gsis.len() > 20 {
        return Err("One or more parameter values were invalid: GlobalSecondaryIndex count exceeds the per-table limit of 20".to_string());
    }

    // PAY_PER_REQUEST billing mode check
    if bm == "PAY_PER_REQUEST" {
        for gsi in gsis {
            if gsi.provisioned_throughput.is_some() {
                return Err(format!(
                    "One or more parameter values were invalid: ProvisionedThroughput should not be specified for index: {} when BillingMode is PAY_PER_REQUEST",
                    gsi.index_name
                ));
            }
        }
    }

    Ok(())
}

fn validate_lsi_structure(
    lsi: &LocalSecondaryIndex,
    table_ks: &[KeySchemaElement],
) -> std::result::Result<(), String> {
    // Key schema structure first
    validate_key_schema_structure(&lsi.key_schema)?;

    // Range key presence (before projection, per DynamoDB ordering)
    let lsi_sk = lsi.key_schema.iter().find(|k| k.key_type == KeyType::RANGE);
    if lsi_sk.is_none() {
        return Err(format!(
            "One or more parameter values were invalid: Index KeySchema does not have a range key for index: {}",
            lsi.index_name
        ));
    }

    // Hash key must match table hash key (before projection)
    let table_pk = table_ks
        .iter()
        .find(|k| k.key_type == KeyType::HASH)
        .map(|k| k.attribute_name.as_str());
    let lsi_pk = lsi
        .key_schema
        .iter()
        .find(|k| k.key_type == KeyType::HASH)
        .map(|k| k.attribute_name.as_str());
    if lsi_pk != table_pk {
        return Err(format!(
            "One or more parameter values were invalid: \
             Index KeySchema does not have the same leading hash key as table KeySchema \
             for index: {}. index hash key: {}, table hash key: {}",
            lsi.index_name,
            lsi_pk.unwrap_or("null"),
            table_pk.unwrap_or("null")
        ));
    }

    // Projection (after range key and hash key checks)
    validate_proj_structure(&lsi.projection)?;

    Ok(())
}

fn validate_gsi_structure(gsi: &GlobalSecondaryIndex) -> std::result::Result<(), String> {
    crate::validation::check_gsi_key_schema(&gsi.key_schema)?;
    validate_proj_structure(&gsi.projection)?;
    Ok(())
}

fn validate_proj_structure(p: &Projection) -> std::result::Result<(), String> {
    // Matched on the projection type regardless of whether NonKeyAttributes is
    // present, so the INCLUDE-without-list case (DynamoDB rejects it) is
    // reachable. ALL and KEYS_ONLY must not carry a list; INCLUDE requires a
    // non-empty one. Reached by both GSI and LSI validation.
    match &p.projection_type {
        None => Err(
            "One or more parameter values were invalid: Unknown ProjectionType: null".to_string(),
        ),
        Some(ProjectionType::ALL) => match &p.non_key_attributes {
            Some(_) => Err("One or more parameter values were invalid: ProjectionType is ALL, but NonKeyAttributes is specified".to_string()),
            None => Ok(()),
        },
        Some(ProjectionType::KEYS_ONLY) => match &p.non_key_attributes {
            Some(_) => Err("One or more parameter values were invalid: ProjectionType is KEYS_ONLY, but NonKeyAttributes is specified".to_string()),
            None => Ok(()),
        },
        Some(ProjectionType::INCLUDE) => match &p.non_key_attributes {
            None => Err("One or more parameter values were invalid: ProjectionType is INCLUDE, but NonKeyAttributes is not specified".to_string()),
            Some(nka) if nka.is_empty() => Err("One or more parameter values were invalid: NonKeyAttributes must not be empty".to_string()),
            Some(_) => Ok(()),
        },
    }
}