resolved.rs

use std::collections::{BTreeMap, HashMap, HashSet};

use indexmap::IndexMap;
use serde::{Deserialize, Serialize};
use thiserror::Error;

use open_dds::{
    data_source::{self, DataConnectorUrl, DataSourceName},
    data_specification::{
        self, ArgumentName, CommandGraphQlDefinition, CommandName, FieldAccess, FieldName,
        FunctionOrProcedure, GraphQlFieldName, GraphQlRootFieldKind, GraphQlTypeName, InbuiltType,
        ModelName, OperatorName, RelationshipName, Role, Type, TypeName, TypeOutputPermission,
    },
};

use ndc_client as ndc;

use super::ndc_validation;

#[derive(Error, Debug)]
pub enum Error {
    #[error("the following data source is defined more than once: {name:}")]
    DuplicateDataSourceDefinition { name: DataSourceName },
    #[error("no data connector available for this data source: {data_source:}")]
    DataConnectorUnavailable { data_source: DataSourceName },
    #[error("the following type is defined more than once: {name:}")]
    DuplicateTypeDefinition { name: TypeName },
    #[error("the following field in type {type_name:} is defined more than once: {field_name:}")]
    DuplicateFieldDefinition {
        type_name: TypeName,
        field_name: FieldName,
    },
    #[error("the following model is defined more than once: {name:}")]
    DuplicateModelDefinition { name: ModelName },
    #[error("the data type {data_type:} for model {model_name:} has not been defined")]
    UnknownModelDataType {
        model_name: ModelName,
        data_type: TypeName,
    },
    #[error(
        "the following argument in model {model_name:} is defined more than once: {argument_name:}"
    )]
    DuplicateModelArgumentDefinition {
        model_name: ModelName,
        argument_name: ArgumentName,
    },
    #[error("unknown model used in model source definition: {model_name:}")]
    UnknownModelInModelSource { model_name: ModelName },
    #[error("source for the following model is defined more than once: {model_name:}")]
    DuplicateModelSourceDefinition { model_name: ModelName },
    #[error("the type {type_name:} referenced in type mappings for model {model_name:} has not been defined")]
    UnknownModelSourceType {
        model_name: ModelName,
        type_name: TypeName,
    },
    #[error("{error:} in model {model_name:}")]
    ModelTypeMappingValidationError {
        model_name: ModelName,
        error: TypeMappingValidationError,
    },
    // ------- Errors for commands ---------
    #[error("the following command is defined more than once: {name:}")]
    DuplicateCommandDefinition { name: CommandName },
    #[error("the following argument {argument_name:} with argument type {argument_type:} in command {command_name:} has not been defined")]
    UnknownCommandArgumentType {
        command_name: CommandName,
        argument_name: ArgumentName,
        argument_type: Type,
    },
    #[error(
        "the following argument in command {command_name:} is defined more than once: {argument_name:}"
    )]
    DuplicateCommandArgumentDefinition {
        command_name: CommandName,
        argument_name: ArgumentName,
    },
    #[error("unknown command used in command source definition: {command_name:}")]
    UnknownCommandInCommandSource { command_name: CommandName },
    #[error("source for the following command is defined more than once: {command_name:}")]
    DuplicateCommandSourceDefinition { command_name: CommandName },
    #[error("the data source {data_source:} for command {command_name:} has not been defined")]
    UnknownCommandDataSource {
        command_name: CommandName,
        data_source: DataSourceName,
    },
    #[error(
        "the mapping for type {type_name:} in command {command_name:} is defined more than once"
    )]
    DuplicateTypeMappingDefinitionInCommandSource {
        command_name: CommandName,
        type_name: TypeName,
    },
    #[error(
        "unknown argument {argument_name:} referenced in argument mappings for command {command_name:}"
    )]
    UnknownCommandSourceArgument {
        command_name: CommandName,
        argument_name: ArgumentName,
    },
    #[error(
        "the mapping for argument {argument_name:} of command {command_name:} has been defined more than once"
    )]
    DuplicateCommandArgumentMapping {
        command_name: CommandName,
        argument_name: ArgumentName,
    },
    #[error("the procedure {procedure:} in the data source {data_source:} for command {command_name:} has not been defined")]
    UnknownCommandProcedure {
        command_name: CommandName,
        data_source: DataSourceName,
        procedure: String,
    },
    #[error("the function {function:} in the data source {data_source:} for command {command_name:} has not been defined")]
    UnknownCommandFunction {
        command_name: CommandName,
        data_source: DataSourceName,
        function: String,
    },
    #[error("{error:} in command {command_name:}")]
    CommandTypeMappingValidationError {
        command_name: CommandName,
        error: TypeMappingValidationError,
    },
    #[error("mismatch between command source and root field type for command {command_name:}")]
    CommandSourceAndRootFieldTypeMismatch { command_name: CommandName },
    // ----------------
    #[error("the mapping for type {type_name:} in model {model_name:} is incompatible with the type representation")]
    IncompatiableTypeMappingDefinition {
        model_name: ModelName,
        type_name: TypeName,
    },
    #[error("the mapping for type {type_name:} in model {model_name:} is defined more than once")]
    DuplicateTypeMappingDefinition {
        model_name: ModelName,
        type_name: TypeName,
    },
    #[error("the data source {data_source:} for model {model_name:} has not been defined")]
    UnknownModelDataSource {
        model_name: ModelName,
        data_source: DataSourceName,
    },
    #[error("the collection {collection:} in the data source {data_source:} for model {model_name:} has not been defined")]
    UnknownModelCollection {
        model_name: ModelName,
        data_source: DataSourceName,
        collection: String,
    },
    #[error(
        "unknown field {field_name:} in field mappings of type {type_name:} in model {model_name:}"
    )]
    UnknownFieldInFieldMapping {
        model_name: ModelName,
        type_name: TypeName,
        field_name: FieldName,
    },
    #[error(
        "the mapping for field {field_name:} of type {type_name:} in model {model_name:} has been defined more than once"
    )]
    DuplicateFieldMapping {
        model_name: ModelName,
        type_name: TypeName,
        field_name: FieldName,
    },
    #[error(
        "unknown argument {argument_name:} referenced in argument mappings for model {model_name:}"
    )]
    UnknownModelSourceArgument {
        model_name: ModelName,
        argument_name: ArgumentName,
    },
    #[error(
        "the mapping for argument {argument_name:} of model {model_name:} has been defined more than once"
    )]
    DuplicateModelArgumentMapping {
        model_name: ModelName,
        argument_name: ArgumentName,
    },
    #[error("model arguments graphql input configuartion has been specified for model {model_name:} that does not have arguments")]
    UnnecessaryModelArgumentsGraphQlInputConfiguration { model_name: ModelName },
    #[error("a graphql type has been defined for an unknown type: {type_name:}")]
    UnknownGraphQlType { type_name: TypeName },
    #[error("multiple graphql types have been defined for type: {type_name:}")]
    DuplicateGraphQlType { type_name: TypeName },
    #[error("multiple graphql types found with the same name: {graphql_type_name:}")]
    ConflictingGraphQlType { graphql_type_name: GraphQlTypeName },
    #[error("unknown model used in graphql definition: {model_name:}")]
    UnknownModelInGraphQlDefinition { model_name: ModelName },
    #[error("unknown command used in graphql definition: {command_name:}")]
    UnknownCommandInGraphQlDefinition { command_name: CommandName },
    #[error("unknown field {field_name:} in unique identifier defined for model {model_name:}")]
    UnknownFieldInUniqueIdentifier {
        model_name: ModelName,
        field_name: FieldName,
    },
    #[error("duplicate field {field_name:} in unique identifier defined for model {model_name:}")]
    DuplicateFieldInUniqueIdentifier {
        model_name: ModelName,
        field_name: FieldName,
    },
    #[error("invalid type represented for model {model_name:}: {type_representation:}")]
    InvalidTypeRepresentation {
        model_name: ModelName,
        type_representation: TypeRepresentation,
    },

    // Permission errors
    #[error("unsupported type in output type permissions definition: {type_name:}; only object types are supported")]
    UnsupportedTypeInOutputPermissions { type_name: TypeName },
    #[error("multiple output type permissions have been defined for type: {type_name:}")]
    DuplicateOutputTypePermissions { type_name: TypeName },
    #[error("unknown type used in output permissions: {type_name:}")]
    UnknownTypeInOutputPermissionsDefinition { type_name: TypeName },
    #[error("unknown field '{field_name:}' used in output permissions of type '{type_name:}'")]
    UnknownFieldInOutputPermissionsDefinition {
        field_name: FieldName,
        type_name: TypeName,
    },
    #[error("unknown model used in model select permissions definition: {model_name:}")]
    UnknownModelInModelSelectPermissions { model_name: ModelName },
    #[error("multiple select permissions defined for model: {model_name:}")]
    DuplicateModelSelectPermission { model_name: ModelName },
    #[error(
        "model source is required for model '{model_name:}' to resolve select permission predicate"
    )]
    ModelSourceRequiredForPredicate { model_name: ModelName },
    #[error("unknown field '{field_name:}' used in select permissions of model '{model_name:}'")]
    UnknownFieldInSelectPermissionsDefinition {
        field_name: FieldName,
        model_name: ModelName,
    },
    #[error("field mapping is missing for field '{field_name:}' used in select permissions of model '{model_name:}'")]
    MissingMappingForSelectPermissionPredicateField {
        field_name: FieldName,
        model_name: ModelName,
    },
    #[error("field '{field_name:}' used in select permissions of model '{model_name:}' should be mapped to non-array scalar field")]
    UnsupportedFieldInSelectPermissionsPredicate {
        field_name: FieldName,
        model_name: ModelName,
    },
    #[error(
        "Invalid operator used in model '{model_name:}' select permission: '{operator_name:}'"
    )]
    InvalidOperator {
        model_name: ModelName,
        operator_name: OperatorName,
    },
    #[error("unknown command used in command permissions definition: {command_name:}")]
    UnknownCommandInCommandPermissions { command_name: CommandName },
    #[error("multiple permissions defined for command: {command_name:}")]
    DuplicateCommandPermission { command_name: CommandName },

    #[error("{message:}")]
    UnsupportedFeature { message: String },
    #[error("the referenced secret {secret_name:} was not found in the environment")]
    SecretNotFound { secret_name: String },
    #[error("{0}")]
    DeserializationError(#[from] serde_json::Error),
    #[error("NDC validation error: {0}")]
    NDCValidationError(#[from] super::ndc_validation::NDCValidationError),
    #[error(
        "duplicate relationship {relationship_name:} associated with source type {type_name:}"
    )]
    DuplicateRelationshipInSourceType {
        type_name: TypeName,
        relationship_name: RelationshipName,
    },
    #[error("unknown model {model_name:} used in relationship {relationship_name:} definition at {unknown_model_source:}")]
    UnknownModelUsedInRelationship {
        relationship_name: RelationshipName,
        model_name: ModelName,
        unknown_model_source: String,
    },
    #[error("Source type {type_name:} referenced in the definition of relationship(s) {relationship_name:} is not defined ")]
    RelationshipDefinedOnUnknownType {
        relationship_name: RelationshipName,
        type_name: TypeName,
    },
    #[error("{reason:}")]
    NotSupported { reason: String },
    #[error("The field path provided in the {location:} of the relationship {relationship_name} is empty")]
    EmptyFieldPath {
        location: String,
        relationship_name: RelationshipName,
    },
    #[error("unknown data source {data_source:} referenced in scalar type representation of {scalar_type:}")]
    ScalarTypeFromUnknownDataSource {
        data_source: DataSourceName,
        scalar_type: String,
    },
    #[error("cannot find scalar type {scalar_type:} in data source {data_source:}")]
    UnknownScalarTypeInDataSource {
        data_source: DataSourceName,
        scalar_type: String,
    },
    #[error("unknown type represented for scalar type {scalar_type:}: {type_name:}")]
    ScalarTypeUnknownRepresentation {
        scalar_type: String,
        type_name: TypeName,
    },
    #[error("multiple type representations defined for scalar {scalar_type:} from data source {data_source:}")]
    DuplicateDataConnectorScalarRepresentation {
        data_source: DataSourceName,
        scalar_type: String,
    },
    #[error("multiple scalar type graphql expression defined for scalar {scalar_type:} from data source {data_source:}")]
    DuplicateScalarGraphQlExpression {
        data_source: DataSourceName,
        scalar_type: String,
    },
    #[error(
        "scalar type representation required for type {scalar_type:} in data source {data_source:}"
    )]
    DataConnectorScalarRepresentationRequired {
        data_source: DataSourceName,
        scalar_type: String,
    },
    #[error("type mapping required for type {type_name:} in model source {model_name:} backed by data source {data_source:}")]
    TypeMappingRequired {
        model_name: ModelName,
        type_name: TypeName,
        data_source: DataSourceName,
    },
    #[error("Unknown field {field_name:} in global_id defined for the type {type_name:}")]
    UnknownFieldInGlobalId {
        field_name: FieldName,
        type_name: TypeName,
    },
    #[error("Model {model_name:} is marked as a global ID source but there are no global id fields present in the related object type {type_name:} ")]
    NoGlobalFieldsPresentInGlobalIdSource {
        type_name: TypeName,
        model_name: ModelName,
    },
    #[error("A field named `id` cannot be present in the object type {type_name} when global_id fields are non-empty.")]
    IdFieldConflictingGlobalId { type_name: TypeName },
    #[error("Found multiple models  {model_1:}, {model_2:} that implement the same object type {object_type:} to be global ID sources.")]
    DuplicateModelGlobalIdSource {
        model_1: ModelName,
        model_2: ModelName,
        object_type: TypeName,
    },
}

#[derive(Error, Debug)]
pub enum TypeMappingValidationError {
    #[error("the type {type_name:} referenced in type mappings has not been defined")]
    UnknownSourceType { type_name: TypeName },
    #[error("the mapping for type {type_name:} is incompatible with the type representation")]
    IncompatiableTypeMappingDefinition { type_name: TypeName },
    #[error("unknown field {field_name:} in field mappings of type {type_name:}")]
    UnknownSourceField {
        type_name: TypeName,
        field_name: FieldName,
    },
    #[error("unknown target column name {column_name:} for field {field_name:} ")]
    UnknownTargetColumn {
        column_name: String,
        field_name: FieldName,
    },
    #[error(
        "the mapping for field {field_name:} of type {type_name:} has been defined more than once"
    )]
    DuplicateFieldMapping {
        type_name: TypeName,
        field_name: FieldName,
    },
}

/// Resolved and validated metadata for a project. Used internally in the v3 server.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct Metadata {
    pub types: HashMap<TypeName, TypeRepresentation>,
    pub models: IndexMap<ModelName, Model>,
    pub commands: IndexMap<CommandName, Command>,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, derive_more::Display)]
pub enum TypeRepresentation {
    #[display(fmt = "Display")]
    Object {
        fields: IndexMap<FieldName, FieldDefinition>,
        relationships: IndexMap<FieldName, Relationship>,
        type_permissions: HashMap<Role, TypeOutputPermission>,
        global_id_fields: Vec<FieldName>,
        graphql_output_type_name: Option<GraphQlTypeName>,
        // TODO: add graphql_output_type_kind if we support creating interfaces.
        // TODO: add graphql_input_type_name if we support complex input types.
    },
    #[display(fmt = "ScalarType")]
    ScalarType {
        graphql_type_name: Option<GraphQlTypeName>,
    },
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct FieldDefinition {
    pub field_type: Type,
}

pub struct ScalarTypeInfo {
    pub scalar_type: ndc::models::ScalarType,
    pub representation: Option<Type>,
    pub comparison_expression_name: Option<GraphQlTypeName>,
}

impl ScalarTypeInfo {
    fn new(source_scalar: ndc::models::ScalarType) -> Self {
        ScalarTypeInfo {
            scalar_type: source_scalar,
            representation: None,
            comparison_expression_name: None,
        }
    }
}

struct DataSourceContext<'a> {
    data_source: &'a data_source::DataSource,
    scalars: HashMap<String, ScalarTypeInfo>,
}

impl<'a> DataSourceContext<'a> {
    fn new(data_source: &'a data_source::DataSource) -> Self {
        DataSourceContext {
            data_source,
            scalars: data_source
                .schema
                .scalar_types
                .clone()
                .into_iter()
                .map(|(k, v)| (k, ScalarTypeInfo::new(v)))
                .collect(),
        }
    }
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct DataSource {
    pub name: DataSourceName,
    pub data_connector_url: DataConnectorUrl,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct ModelArgumentsGraphQlInput {
    pub field: GraphQlFieldName,
    pub graphql_type: GraphQlTypeName,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default)]
pub struct ModelGraphQlApi {
    pub arguments_input: Option<ModelArgumentsGraphQlInput>,
    pub select_uniques: Vec<SelectUniqueGraphQlDefinition>,
    pub select_many: Option<SelectManyGraphQlDefinition>,
    pub filter_expression: Option<ModelFilterExpression>,
    pub order_by_expression: Option<ModelOrderByExpression>,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct Model {
    pub name: ModelName,
    pub data_type: TypeName,
    pub type_representation: TypeRepresentation,
    pub arguments: IndexMap<ArgumentName, Type>,
    pub graphql_api: ModelGraphQlApi,
    pub source: Option<ModelSource>,
    pub select_permissions: Option<HashMap<Role, SelectPermission>>,
    pub global_id_source: bool,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct ModelSource {
    pub data_source: DataSource,
    pub collection: String,
    pub type_mappings: HashMap<TypeName, TypeMapping>,
    pub argument_mappings: HashMap<ArgumentName, String>,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub enum TypeMapping {
    Object {
        field_mappings: HashMap<FieldName, FieldMapping>,
    },
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct FieldMapping {
    pub column: String,
    pub column_type: Option<ndc::models::Type>,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct SelectUniqueGraphQlDefinition {
    pub query_root_field: GraphQlFieldName,
    pub unique_identifier: IndexMap<FieldName, Type>,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct ComparisonExpressionInfo {
    pub data_source_name: DataSourceName,
    pub scalar_type_name: String,
    pub type_name: GraphQlTypeName,
    pub ndc_column: String,
    pub operators: BTreeMap<String, Type>,
}

// TODO: add support for aggregates
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct OrderByExpressionInfo {
    pub ndc_column: String,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct ModelFilterExpression {
    pub where_type_name: GraphQlTypeName,
    pub scalar_fields: HashMap<FieldName, ComparisonExpressionInfo>,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct ModelOrderByExpression {
    pub data_source_name: DataSourceName,
    pub order_by_type_name: GraphQlTypeName,
    pub order_by_fields: HashMap<FieldName, OrderByExpressionInfo>,
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct SelectManyGraphQlDefinition {
    pub query_root_field: GraphQlFieldName,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub enum RelationshipTarget {
    Model {
        model: Model,
        relationship_type: data_specification::RelationshipType,
    },
    // TODO: Add support for relationships with Commands.
    // Command {
    //     command: data_specification::CommandName,
    // },
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct FieldPath {
    /// The path to the `tail`
    pub head: Vec<FieldAccess>,
    /// The last element in the `FieldPath`
    pub tail: FieldAccess,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub enum RelationshipMappingSource {
    Value(data_specification::ValueExpression),
    FieldPath(FieldPath),
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub enum RelationshipMappingTarget {
    Argument(data_specification::ArgumentName),
    ModelField(FieldPath),
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct RelationshipMapping {
    pub source: RelationshipMappingSource,
    pub target: RelationshipMappingTarget,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct Relationship {
    pub name: RelationshipName,
    pub source: TypeName,
    pub target: RelationshipTarget,
    pub target_typename: TypeName,
    pub mappings: Vec<RelationshipMapping>,
    pub relationship_type: data_specification::RelationshipType,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct CommandGraphQlApi {
    pub root_field_kind: GraphQlRootFieldKind,
    pub root_field_name: GraphQlFieldName,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct Command {
    pub name: CommandName,
    pub output_type: Type,
    pub arguments: IndexMap<ArgumentName, Type>,
    pub graphql_api: Option<CommandGraphQlDefinition>,
    pub source: Option<CommandSource>,
    pub permissions: Option<HashMap<Role, CommandPermission>>,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct CommandSource {
    pub data_source: DataSource,
    pub source: FunctionOrProcedure,
    pub type_mappings: HashMap<TypeName, TypeMapping>,
    pub argument_mappings: HashMap<ArgumentName, String>,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct CommandPermission {
    pub allow_execution: bool,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct SelectPermission {
    // Missing filter implies all rows are selectable
    pub filter: Option<ModelPredicate>,
    // pub allow_aggregations: bool,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub enum ModelPredicate {
    UnaryFieldComparison {
        field: FieldName,
        ndc_column: String,
        operator: ndc_client::models::UnaryComparisonOperator,
    },
    BinaryFieldComparison {
        field: FieldName,
        ndc_column: String,
        operator: ndc_client::models::BinaryComparisonOperator,
        argument_type: Type,
        value: data_specification::ValueExpression,
    },
    // TODO: Remote relationships are disallowed for now
    Relationship {
        name: RelationshipName,
        predicate: Option<Box<ModelPredicate>>,
    },

    And(Vec<ModelPredicate>),
    Or(Vec<ModelPredicate>),
    Not(Box<ModelPredicate>),
}

/*******************
    Functions to validate and convert Open DDS spec to internal metadata
*******************/
pub fn resolve_metadata(metadata: &open_dds::Metadata) -> Result<Metadata, Error> {
    let metadata_accessor = open_dds::accessor::MetadataAccessor::new(metadata);

    let mut data_sources = HashMap::new();
    for data_source in &metadata_accessor.data_sources {
        if data_sources
            .insert(
                data_source.name.clone(),
                DataSourceContext::new(data_source),
            )
            .is_some()
        {
            return Err(Error::DuplicateDataSourceDefinition {
                name: data_source.name.clone(),
            });
        }
    }

    let mut types: HashMap<TypeName, TypeRepresentation> = HashMap::new();
    let mut existing_graphql_types: HashSet<GraphQlTypeName> = HashSet::new();
    for object_type_definition in &metadata_accessor.object_types {
        if types
            .insert(
                object_type_definition.name.clone(),
                resolve_object_type(object_type_definition, &mut existing_graphql_types)?,
            )
            .is_some()
        {
            return Err(Error::DuplicateTypeDefinition {
                name: object_type_definition.name.clone(),
            });
        }
    }

    for scalar_type in &metadata_accessor.scalar_types {
        if types
            .insert(
                scalar_type.name.clone(),
                TypeRepresentation::ScalarType {
                    graphql_type_name: scalar_type.graphql.as_ref().map(|c| c.type_name.clone()),
                },
            )
            .is_some()
        {
            return Err(Error::DuplicateTypeDefinition {
                name: scalar_type.name.clone(),
            });
        }
        check_conflicting_graphql_types(
            &mut existing_graphql_types,
            scalar_type.graphql.as_ref().map(|c| &c.type_name),
        )?;
    }

    for output_type_permission in &metadata_accessor.output_permissions {
        match types.get_mut(&output_type_permission.type_name) {
            None => {
                return Err(Error::UnknownTypeInOutputPermissionsDefinition {
                    type_name: output_type_permission.type_name.clone(),
                })
            }
            Some(object_type) => {
                resolve_output_type_permission(object_type, output_type_permission)?;
            }
        }
    }

    for scalar_type_representation in &metadata_accessor.data_connector_scalar_representations {
        let scalar_type_name: &String = &scalar_type_representation.scalar_type;
        let source_context = data_sources
            .get_mut(&scalar_type_representation.data_source)
            .ok_or_else(|| Error::ScalarTypeFromUnknownDataSource {
                scalar_type: scalar_type_name.clone(),
                data_source: scalar_type_representation.data_source.clone(),
            })?;

        let scalar_type = source_context
            .scalars
            .get_mut(&scalar_type_representation.scalar_type)
            .ok_or_else(|| Error::UnknownScalarTypeInDataSource {
                scalar_type: scalar_type_name.clone(),
                data_source: scalar_type_representation.data_source.clone(),
            })?;

        if scalar_type.representation.is_none() {
            match &scalar_type_representation.representation {
                Type::Inbuilt(_) => {} // TODO: Validate Nullable and Array types in Inbuilt
                Type::Custom(type_name) => {
                    let _representation = types.get(type_name).ok_or_else(|| {
                        Error::ScalarTypeUnknownRepresentation {
                            scalar_type: scalar_type_name.clone(),
                            type_name: type_name.clone(),
                        }
                    })?;
                }
            }
            scalar_type.representation = Some(scalar_type_representation.representation.clone());
        } else {
            return Err(Error::DuplicateDataConnectorScalarRepresentation {
                data_source: scalar_type_representation.data_source.clone(),
                scalar_type: scalar_type_name.clone(),
            });
        }
        scalar_type.comparison_expression_name = scalar_type_representation
            .graphql
            .as_ref()
            .and_then(|c| c.comparison_expression_type_name.clone());
        check_conflicting_graphql_types(
            &mut existing_graphql_types,
            scalar_type.comparison_expression_name.as_ref(),
        )?;
    }

    // TODO: validate types

    let mut models = IndexMap::new();
    let mut global_id_models = HashMap::new();

    for model in &metadata_accessor.models {
        let mut resolved_model = resolve_model(model, &types)?;
        if resolved_model.global_id_source {
            match global_id_models.insert(model.data_type.clone(), model.name.clone()) {
                None => {}
                Some(duplicate_model_name) => {
                    return Err(Error::DuplicateModelGlobalIdSource {
                        model_1: resolved_model.name,
                        model_2: duplicate_model_name,
                        object_type: resolved_model.data_type,
                    })
                }
            }
        }
        if let Some(model_source) = &model.source {
            resolve_model_source(model_source, &mut resolved_model, &data_sources, &types)?;
        }
        if let Some(model_graphql_definition) = &model.graphql {
            resolve_model_graphql_api(
                model_graphql_definition,
                &mut resolved_model,
                &mut existing_graphql_types,
                &data_sources,
            )?;
        }
        if models.insert(model.name.clone(), resolved_model).is_some() {
            return Err(Error::DuplicateModelDefinition {
                name: model.name.clone(),
            });
        }
    }

    for relationship in &metadata_accessor.relationships {
        let resolved_relationship = resolve_relationship(relationship, &models)?;
        let source_type = types.get_mut(&relationship.source).ok_or_else(|| {
            Error::RelationshipDefinedOnUnknownType {
                relationship_name: relationship.name.clone(),
                type_name: relationship.source.to_owned(),
            }
        })?;
        let relationship_field_name = FieldName(relationship.name.0.clone());

        match source_type {
            TypeRepresentation::Object {
                ref mut relationships,
                ..
            } => {
                if relationships
                    .insert(relationship_field_name, resolved_relationship)
                    .is_some()
                {
                    return Err(Error::DuplicateRelationshipInSourceType {
                        type_name: relationship.source.clone(),
                        relationship_name: relationship.name.clone(),
                    });
                }
            }
            TypeRepresentation::ScalarType { .. } => {
                return Err(Error::NotSupported {
                    reason: "A relationship can only be defined on an OBJECT type.".to_string(),
                })
            }
        }
    }

    // resolve all the command without their sources
    let mut commands = IndexMap::new();
    for command in &metadata_accessor.commands {
        let mut resolved_command = resolve_command(command, &types)?;
        if let Some(command_source) = &command.source {
            resolve_command_source(command_source, &mut resolved_command, &data_sources, &types)?;
        }
        if commands
            .insert(command.name.clone(), resolved_command)
            .is_some()
        {
            return Err(Error::DuplicateCommandDefinition {
                name: command.name.clone(),
            });
        }
    }

    for command_permissions in &metadata_accessor.command_permissions {
        let command_name = &command_permissions.command_name;
        let command = commands.get_mut(command_name).ok_or_else(|| {
            Error::UnknownCommandInCommandPermissions {
                command_name: command_name.clone(),
            }
        })?;
        if command.permissions.is_none() {
            command.permissions = Some(resolve_command_permissions(command_permissions)?);
        } else {
            return Err(Error::DuplicateCommandPermission {
                command_name: command_name.clone(),
            });
        }
    }

    for select_permissions in &metadata_accessor.select_permissions {
        let model_name = &select_permissions.model_name;
        let model = models.get_mut(model_name).ok_or_else(|| {
            Error::UnknownModelInModelSelectPermissions {
                model_name: model_name.clone(),
            }
        })?;
        if model.select_permissions.is_none() {
            model.select_permissions = Some(resolve_model_select_permissions(
                model,
                select_permissions,
                &data_sources,
            )?);
        } else {
            return Err(Error::DuplicateModelSelectPermission {
                model_name: model_name.clone(),
            });
        }
    }

    Ok(Metadata {
        types,
        models,
        commands,
    })
}

fn check_conflicting_graphql_types(
    existing_graphql_types: &mut HashSet<GraphQlTypeName>,
    new_graphql_type: Option<&GraphQlTypeName>,
) -> Result<(), Error> {
    if let Some(new_graphql_type) = new_graphql_type {
        // Fail on conflicting graphql type names
        if !(existing_graphql_types.insert(new_graphql_type.clone())) {
            return Err(Error::ConflictingGraphQlType {
                graphql_type_name: new_graphql_type.clone(),
            });
        }
    }
    Ok(())
}

fn resolve_object_type(
    object_type_definition: &data_specification::ObjectDefinition,
    existing_graphql_types: &mut HashSet<GraphQlTypeName>,
) -> Result<TypeRepresentation, Error> {
    let mut resolved_fields = IndexMap::new();
    let mut resolved_global_id_fields = Vec::new();

    for field in &object_type_definition.fields {
        if resolved_fields
            .insert(field.name.clone(), resolve_field(field)?)
            .is_some()
        {
            return Err(Error::DuplicateFieldDefinition {
                type_name: object_type_definition.name.clone(),
                field_name: field.name.clone(),
            });
        }
    }
    match &object_type_definition.global_id {
        Some(global_id_fields) => {
            // Throw error if the object type has a field called id" and has global fields configured.
            // Because, when the global id fields are configured, the `id` field will be auto-generated.
            if !global_id_fields.is_empty() && resolved_fields.contains_key(&FieldName("id".into()))
            {
                return Err(Error::IdFieldConflictingGlobalId {
                    type_name: object_type_definition.name.clone(),
                });
            };
            for global_id_field in global_id_fields {
                if !resolved_fields.contains_key(global_id_field) {
                    return Err(Error::UnknownFieldInGlobalId {
                        field_name: global_id_field.clone(),
                        type_name: object_type_definition.name.clone(),
                    });
                } else {
                    resolved_global_id_fields.push(global_id_field.clone())
                }
            }
        }
        None => {}
    }
    check_conflicting_graphql_types(
        existing_graphql_types,
        object_type_definition
            .graphql
            .as_ref()
            .and_then(|c| c.type_name.as_ref()),
    )?;
    Ok(TypeRepresentation::Object {
        fields: resolved_fields,
        relationships: IndexMap::new(),
        global_id_fields: resolved_global_id_fields,
        type_permissions: HashMap::new(),
        graphql_output_type_name: object_type_definition
            .graphql
            .as_ref()
            .and_then(|c| c.type_name.clone()),
    })
}

fn resolve_field(
    field: &open_dds::data_specification::FieldDefinition,
) -> Result<FieldDefinition, Error> {
    Ok(FieldDefinition {
        field_type: field.field_type.clone(),
    })
}

fn resolve_model(
    model: &open_dds::data_specification::Model,
    types: &HashMap<TypeName, TypeRepresentation>,
) -> Result<Model, Error> {
    let model_type_representation =
        types
            .get(&model.data_type)
            .ok_or_else(|| Error::UnknownModelDataType {
                model_name: model.name.clone(),
                data_type: model.data_type.clone(),
            })?;

    let mut global_id_source = false;
    if model.global_id_source {
        if let TypeRepresentation::Object {
            global_id_fields, ..
        } = model_type_representation
        {
            // Check if there are any global fields present in the related
            // object type, if the model is marked as a global source.
            if global_id_fields.is_empty() {
                return Err(Error::NoGlobalFieldsPresentInGlobalIdSource {
                    type_name: model.data_type.clone(),
                    model_name: model.name.clone(),
                });
            } else {
                global_id_source = true;
            };
        }
    };

    let mut arguments = IndexMap::new();
    for argument in &model.arguments {
        if arguments
            .insert(argument.name.clone(), argument.argument_type.clone())
            .is_some()
        {
            return Err(Error::DuplicateModelArgumentDefinition {
                model_name: model.name.clone(),
                argument_name: argument.name.clone(),
            });
        }
    }

    Ok(Model {
        name: model.name.clone(),
        data_type: model.data_type.clone(),
        type_representation: model_type_representation.clone(),
        arguments,
        graphql_api: ModelGraphQlApi::default(),
        source: None,
        select_permissions: None,
        global_id_source,
    })
}

fn is_valid_type(type_obj: &Type, types: &HashMap<TypeName, TypeRepresentation>) -> bool {
    match type_obj {
        Type::Inbuilt(inbuilt_type) => match inbuilt_type {
            InbuiltType::Nullable(nullable_type) => is_valid_type(nullable_type, types),
            InbuiltType::Array(array_type) => is_valid_type(array_type, types),
            _ => true,
        },
        Type::Custom(type_name) => types.get(type_name).is_some(),
    }
}

fn resolve_command(
    command: &open_dds::data_specification::Command,
    types: &HashMap<TypeName, TypeRepresentation>,
) -> Result<Command, Error> {
    let mut arguments = IndexMap::new();
    // duplicate command arguments should not be allowed
    for argument in &command.arguments {
        if is_valid_type(&argument.argument_type, types) {
            if arguments
                .insert(argument.name.clone(), argument.argument_type.clone())
                .is_some()
            {
                return Err(Error::DuplicateCommandArgumentDefinition {
                    command_name: command.name.clone(),
                    argument_name: argument.name.clone(),
                });
            }
        } else {
            return Err(Error::UnknownCommandArgumentType {
                command_name: command.name.clone(),
                argument_name: argument.name.clone(),
                argument_type: argument.argument_type.clone(),
            });
        }
    }

    Ok(Command {
        name: command.name.clone(),
        output_type: command.output_type.clone(),
        arguments,
        graphql_api: command.graphql.clone(),
        source: None,
        permissions: None,
    })
}

fn resolve_command_source(
    command_source: &open_dds::data_specification::CommandSource,
    command: &mut Command,
    data_sources: &HashMap<DataSourceName, DataSourceContext>,
    types: &HashMap<TypeName, TypeRepresentation>,
) -> Result<(), Error> {
    if command.source.is_some() {
        return Err(Error::DuplicateCommandSourceDefinition {
            command_name: command.name.clone(),
        });
    }

    // check if data_source for the command exists
    let data_source_context = data_sources
        .get(&command_source.data_source_name)
        .ok_or_else(|| Error::UnknownCommandDataSource {
            command_name: command.name.clone(),
            data_source: command_source.data_source_name.clone(),
        })?;

    // TODO: Add type mappings for complex input types
    // Construct the type_mappings when command.output_type is a Custom type
    let type_mappings = match &command.output_type {
        Type::Inbuilt(_inbuilt_type) => HashMap::new(),
        Type::Custom(custom_type_name) => {
            // Get either the result type of function or procedure used as the ndc source for commands
            let source_result_type = match &command_source.source.0 {
                FunctionOrProcedure::Procedure(procedure) => {
                    let source_procedure = data_source_context
                        .data_source
                        .schema
                        .procedures
                        .iter()
                        .find(|proc| proc.name == *procedure)
                        .ok_or_else(|| Error::UnknownCommandProcedure {
                            command_name: command.name.clone(),
                            data_source: data_source_context.data_source.name.clone(),
                            procedure: procedure.clone(),
                        })?;

                    source_procedure.result_type.clone()
                }
                FunctionOrProcedure::Function(function) => {
                    let source_function = data_source_context
                        .data_source
                        .schema
                        .functions
                        .iter()
                        .find(|func| func.name == *function)
                        .ok_or_else(|| Error::UnknownCommandFunction {
                            command_name: command.name.clone(),
                            data_source: data_source_context.data_source.name.clone(),
                            function: function.clone(),
                        })?;

                    source_function.result_type.clone()
                }
            };

            let mut type_mappings = HashMap::new();

            // Get the corresponding object_type (data_source.object_type) associated with the result_type for the source
            let source_result_type_name =
                ndc_validation::get_result_type_func_proc(&source_result_type);
            let source_result_object_type = data_source_context
                .data_source
                .schema
                .object_types
                .get(source_result_type_name)
                .ok_or(ndc_validation::NDCValidationError::NoSuchType(
                    source_result_type_name.to_string(),
                ))?;

            // Resolve the type mappings between the source result_type and the type_mappings present in command_source
            for (type_name, type_mapping) in &command_source.type_mappings {
                let resolved_type_mapping =
                    resolve_type_mapping(type_name, type_mapping, types, source_result_object_type)
                        .map_err(|type_validation_error| {
                            Error::CommandTypeMappingValidationError {
                                command_name: command.name.clone(),
                                error: type_validation_error,
                            }
                        })?;

                if type_mappings
                    .insert(type_name.clone(), resolved_type_mapping)
                    .is_some()
                {
                    return Err(Error::DuplicateTypeMappingDefinitionInCommandSource {
                        command_name: command.name.clone(),
                        type_name: type_name.clone(),
                    });
                }
            }

            // Ensure that any unmapped types get implicitly mapped.
            //For eg: if the command output_type does not have type_mappings,
            // implicity assume that the field of output_type maps with the same
            // name
            add_implicit_type_mappings(
                custom_type_name,
                types,
                &mut type_mappings,
                source_result_object_type,
            )
            .map_err(|type_validation_error| {
                Error::CommandTypeMappingValidationError {
                    command_name: command.name.clone(),
                    error: type_validation_error,
                }
            })?;
            type_mappings
        }
    };

    let mut argument_mappings = HashMap::new();
    for argument_mapping in &command_source.argument_mappings {
        // Ensure that the argument being mapped actually exists
        if !command.arguments.contains_key(argument_mapping.0) {
            return Err(Error::UnknownCommandSourceArgument {
                command_name: command.name.clone(),
                argument_name: argument_mapping.0.clone(),
            });
        }

        if argument_mappings
            .insert(argument_mapping.0.clone(), argument_mapping.1.clone())
            .is_some()
        {
            return Err(Error::DuplicateCommandArgumentMapping {
                command_name: command.name.clone(),
                argument_name: argument_mapping.0.clone(),
            });
        }
    }

    for argument_name in command.arguments.keys() {
        // If no mapping is defined for an argument, implicitly assume it maps with
        // the same name
        if !argument_mappings.contains_key(argument_name) {
            argument_mappings.insert(argument_name.clone(), argument_name.0.clone());
        }
    }

    command.source = Some(CommandSource {
        data_source: DataSource {
            name: command_source.data_source_name.clone(),
            data_connector_url: data_source_context.data_source.data_connector_url.clone(),
        },
        source: command_source.source.0.clone(),
        type_mappings,
        argument_mappings,
    });

    ndc_validation::validate_ndc_command(
        &command.name,
        command,
        &data_source_context.data_source.schema,
    )?;

    Ok(())
}

fn resolve_command_permissions(
    permissions: &open_dds::data_specification::CommandPermissions,
) -> Result<HashMap<Role, CommandPermission>, Error> {
    let mut validated_permissions = HashMap::new();
    for (role, permission) in &permissions.permissions {
        // TODO: Use the permission predicates/presets
        let resolved_permission = CommandPermission {
            allow_execution: permission.allow_execution,
        };
        validated_permissions.insert(role.clone(), resolved_permission);
    }
    Ok(validated_permissions)
}

fn resolve_model_select_permissions(
    model: &Model,
    select_permissions: &open_dds::data_specification::ModelSelectPermissions,
    data_sources: &HashMap<DataSourceName, DataSourceContext>,
) -> Result<HashMap<Role, SelectPermission>, Error> {
    if let TypeRepresentation::Object { fields, .. } = &model.type_representation {
        let mut validated_permissions = HashMap::new();
        for (role, permission) in &select_permissions.permissions {
            let resolved_predicate = &permission
                .filter
                .as_ref()
                .map(|model_predicate| {
                    resolve_model_predicate(model_predicate, model, data_sources, fields)
                })
                .transpose()?;
            let resolved_permission = SelectPermission {
                filter: resolved_predicate.clone(),
            };
            validated_permissions.insert(role.clone(), resolved_permission);
        }
        Ok(validated_permissions)
    } else {
        Err(Error::InvalidTypeRepresentation {
            model_name: model.name.clone(),
            type_representation: model.type_representation.clone(),
        })
    }
}

fn resolve_model_predicate(
    model_predicate: &data_specification::ModelPredicate,
    model: &Model,
    data_sources: &HashMap<DataSourceName, DataSourceContext>,
    fields: &IndexMap<FieldName, FieldDefinition>,
    // type_representation: &TypeRepresentation,
) -> Result<ModelPredicate, Error> {
    match model_predicate {
        data_specification::ModelPredicate::FieldComparison {
            field,
            operator,
            value,
        } => {
            // TODO: (anon) typecheck the value expression with the field
            // TODO: resolve the "in" operator too (ndc_client::models::BinaryArrayComparisonOperator)
            if let Some(model_source) = &model.source {
                // Get field mappings of model data type
                let TypeMapping::Object { field_mappings } = model_source
                    .type_mappings
                    .get(&model.data_type)
                    .ok_or(Error::TypeMappingRequired {
                        model_name: model.name.clone(),
                        type_name: model.data_type.clone(),
                        data_source: model_source.data_source.name.clone(),
                    })?;
                // Determine field_mapping for the predicate field
                let field_mapping = field_mappings.get(field).ok_or_else(|| {
                    Error::UnknownFieldInSelectPermissionsDefinition {
                        field_name: field.clone(),
                        model_name: model.name.clone(),
                    }
                })?;
                // Determine ndc type of the field
                let field_ndc_type = field_mapping.column_type.as_ref().ok_or_else(|| {
                    Error::MissingMappingForSelectPermissionPredicateField {
                        field_name: field.clone(),
                        model_name: model.name.clone(),
                    }
                })?;
                // Determine whether the ndc type is a simple scalar
                let field_ndc_type_scalar =
                    get_simple_scalar(field_ndc_type.clone()).ok_or_else(|| {
                        Error::UnsupportedFieldInSelectPermissionsPredicate {
                            field_name: field.clone(),
                            model_name: model.name.clone(),
                        }
                    })?;
                // Get available scalars defined in the data source
                let scalars = &data_sources
                    .get(&model_source.data_source.name)
                    .ok_or(Error::UnknownModelDataSource {
                        model_name: model.name.clone(),
                        data_source: model_source.data_source.name.clone(),
                    })?
                    .scalars;
                // Get scalar type info from the data source
                let scalar_type_info = scalars.get(&field_ndc_type_scalar).ok_or_else(|| {
                    Error::UnknownScalarTypeInDataSource {
                        scalar_type: field_ndc_type_scalar.clone(),
                        data_source: model_source.data_source.name.clone(),
                    }
                })?;

                if let Some(value_expression) = value {
                    let (resolved_operator, argument_type) = resolve_binary_operator(
                        operator,
                        &model.name,
                        &model_source.data_source.name,
                        field,
                        fields,
                        scalars,
                        &scalar_type_info.scalar_type,
                    )?;
                    Ok(ModelPredicate::BinaryFieldComparison {
                        field: field.clone(),
                        ndc_column: field_mapping.column.clone(),
                        operator: resolved_operator,
                        argument_type,
                        value: value_expression.clone(),
                    })
                } else {
                    Ok(ModelPredicate::UnaryFieldComparison {
                        field: field.clone(),
                        ndc_column: field_mapping.column.clone(),
                        operator: resolve_unary_operator(&model.name, operator)?,
                    })
                }
            } else {
                Err(Error::ModelSourceRequiredForPredicate {
                    model_name: model.name.clone(),
                })
            }
        }
        data_specification::ModelPredicate::Relationship {
            name: _,
            predicate: _,
        } => {
            Err(Error::UnsupportedFeature {
                message: "'relationship' model predicate is not supported yet.".into(),
            })
            /* TODO: uncomment when we suppor this
            if let Some(nested_predicate) = predicate {
                let resolved_predicate =
                    resolve_model_predicate(nested_predicate, model_name, type_representation)?;
                Ok(ModelPredicate::Relationship {
                    name: name.clone(),
                    predicate: Some(Box::new(resolved_predicate)),
                })
            } else {
                Ok(ModelPredicate::Relationship {
                    name: name.clone(),
                    predicate: None,
                })
            }
            */
        }
        data_specification::ModelPredicate::Not(predicate) => {
            let resolved_predicate =
                resolve_model_predicate(predicate, model, data_sources, fields)?;
            Ok(ModelPredicate::Not(Box::new(resolved_predicate)))
        }
        data_specification::ModelPredicate::And(predicates) => {
            let mut resolved_predicates = Vec::new();
            for predicate in predicates {
                resolved_predicates.push(resolve_model_predicate(
                    predicate,
                    model,
                    data_sources,
                    fields,
                )?);
            }
            Ok(ModelPredicate::And(resolved_predicates))
        }
        data_specification::ModelPredicate::Or(predicates) => {
            let mut resolved_predicates = Vec::new();
            for predicate in predicates {
                resolved_predicates.push(resolve_model_predicate(
                    predicate,
                    model,
                    data_sources,
                    fields,
                )?);
            }
            Ok(ModelPredicate::Or(resolved_predicates))
        }
    }
}

fn resolve_binary_operator(
    operator: &OperatorName,
    model_name: &ModelName,
    data_source: &DataSourceName,
    field_name: &FieldName,
    fields: &IndexMap<FieldName, FieldDefinition>,
    scalars: &HashMap<String, ScalarTypeInfo>,
    ndc_scalar_type: &ndc_client::models::ScalarType,
) -> Result<(ndc_client::models::BinaryComparisonOperator, Type), Error> {
    match operator.0.as_str() {
        "_eq" => {
            let field_definition = fields.get(field_name).ok_or_else(|| {
                Error::UnknownFieldInSelectPermissionsDefinition {
                    field_name: field_name.clone(),
                    model_name: model_name.clone(),
                }
            })?;
            Ok((
                ndc_client::models::BinaryComparisonOperator::Equal,
                field_definition.field_type.clone(),
            ))
        }
        _ => {
            let argument_ndc_type = &ndc_scalar_type
                .comparison_operators
                .get(&operator.0)
                .ok_or_else(|| Error::InvalidOperator {
                    model_name: model_name.clone(),
                    operator_name: operator.clone(),
                })?
                .argument_type;
            Ok((
                ndc_client::models::BinaryComparisonOperator::Other {
                    name: operator.0.clone(),
                },
                resolve_ndc_type(data_source, argument_ndc_type.clone(), scalars)?,
            ))
        }
    }
}

fn resolve_unary_operator(
    model_name: &ModelName,
    operator: &OperatorName,
) -> Result<ndc_client::models::UnaryComparisonOperator, Error> {
    match operator.0.as_str() {
        "_is_null" => Ok(ndc_client::models::UnaryComparisonOperator::IsNull),
        _ => Err(Error::InvalidOperator {
            model_name: model_name.clone(),
            operator_name: operator.clone(),
        }),
    }
}

fn resolve_relationship(
    relationship: &data_specification::Relationship,
    models: &IndexMap<ModelName, Model>,
) -> Result<Relationship, Error> {
    let (relationship_target, target_typename, relationship_type) = match &relationship.target {
        data_specification::RelationshipTarget::Model {
            model_name,
            relationship_type,
        } => {
            let target_model =
                models
                    .get(model_name)
                    .ok_or_else(|| Error::UnknownModelUsedInRelationship {
                        relationship_name: relationship.name.clone(),
                        model_name: model_name.clone(),
                        unknown_model_source: "target".to_string(),
                    })?;
            (
                RelationshipTarget::Model {
                    model: target_model.clone(),
                    relationship_type: relationship_type.clone(),
                },
                target_model.data_type.clone(),
                relationship_type,
            )
        }
    };
    let mut resolved_relationship_mappings = Vec::new();
    for relationship_mapping in &relationship.mappings {
        let resolved_relationship_source_mapping = match &relationship_mapping.source {
            data_specification::RelationshipMappingSource::Value(v) => {
                RelationshipMappingSource::Value(v.clone())
            }
            data_specification::RelationshipMappingSource::FieldPath(field_path) => {
                match &field_path[..] {
                    [] => {
                        return Err(Error::EmptyFieldPath {
                            location: "source".to_string(),
                            relationship_name: relationship.name.clone(),
                        })
                    }
                    [t] => RelationshipMappingSource::FieldPath(FieldPath {
                        head: Vec::new(),
                        tail: t.clone(),
                    }),
                    _ => {
                        return Err(Error::NotSupported {
                            reason: "Relationships with nested field paths are not supported yet."
                                .to_string(),
                        })
                    }
                }
            }
        };
        let resolved_relationship_target_mapping = match &relationship_mapping.target {
            data_specification::RelationshipMappingTarget::Argument(argument_name) => {
                RelationshipMappingTarget::Argument(argument_name.clone())
            }
            data_specification::RelationshipMappingTarget::ModelField(field_path) => {
                match &field_path[..] {
                    [] => {
                        return Err(Error::EmptyFieldPath {
                            location: "source".to_string(),
                            relationship_name: relationship.name.clone(),
                        })
                    }
                    [t] => RelationshipMappingTarget::ModelField(FieldPath {
                        head: Vec::new(),
                        tail: t.clone(),
                    }),
                    _ => {
                        return Err(Error::NotSupported {
                            reason: "Relationships with nested field paths are not supported yet."
                                .to_string(),
                        })
                    }
                }
            }
        };
        let resolved_relationship_mapping = RelationshipMapping {
            source: resolved_relationship_source_mapping,
            target: resolved_relationship_target_mapping,
        };
        resolved_relationship_mappings.push(resolved_relationship_mapping);
    }
    Ok(Relationship {
        name: relationship.name.clone(),
        source: relationship.source.clone(),
        target: relationship_target,
        mappings: resolved_relationship_mappings,
        target_typename,
        relationship_type: relationship_type.clone(),
    })
}

fn resolve_type_mapping(
    type_name: &TypeName,
    type_mapping: &data_specification::TypeMapping,
    types: &HashMap<TypeName, TypeRepresentation>,
    source_collection_type: &ndc::models::ObjectType,
) -> Result<TypeMapping, TypeMappingValidationError> {
    let type_representation =
        types
            .get(type_name)
            .ok_or_else(|| TypeMappingValidationError::UnknownSourceType {
                type_name: type_name.clone(),
            })?;
    match type_mapping {
        data_specification::TypeMapping::Object { field_mappings } => {
            let fields = if let TypeRepresentation::Object { fields, .. } = type_representation {
                fields
            } else {
                return Err(
                    TypeMappingValidationError::IncompatiableTypeMappingDefinition {
                        type_name: type_name.clone(),
                    },
                );
            };

            let mut resolved_field_mappings = HashMap::new();
            for (field_name, field_mapping) in field_mappings {
                // Make sure every field mapping defined corresponds to an actual field
                if !fields.contains_key(field_name) {
                    return Err(TypeMappingValidationError::UnknownSourceField {
                        type_name: type_name.clone(),
                        field_name: field_name.clone(),
                    });
                }
                let source_column = source_collection_type
                    .fields
                    .get(&field_mapping.column)
                    .ok_or(TypeMappingValidationError::UnknownTargetColumn {
                        field_name: field_name.clone(),
                        column_name: field_mapping.column.clone(),
                    })?;
                if resolved_field_mappings
                    .insert(
                        field_name.clone(),
                        FieldMapping {
                            column: field_mapping.column.clone(),
                            column_type: Some(source_column.r#type.clone()),
                        },
                    )
                    .is_some()
                {
                    return Err(TypeMappingValidationError::DuplicateFieldMapping {
                        type_name: type_name.clone(),
                        field_name: field_name.clone(),
                    });
                };
            }

            for field_name in fields.keys() {
                // If no mapping is defined for a field, implicitly create a mapping
                // with the same column name as the field.
                if !resolved_field_mappings.contains_key(field_name) {
                    resolved_field_mappings.insert(
                        field_name.clone(),
                        FieldMapping {
                            column: field_name.0.clone(),
                            column_type: None,
                        },
                    );
                }
            }

            Ok(TypeMapping::Object {
                field_mappings: resolved_field_mappings,
            })
        }
    }
}

fn add_implicit_type_mappings(
    type_name: &TypeName,
    types: &HashMap<TypeName, TypeRepresentation>,
    type_mappings: &mut HashMap<TypeName, TypeMapping>,
    source_collection_type: &ndc::models::ObjectType,
) -> Result<(), TypeMappingValidationError> {
    // Already has a type mapping
    if type_mappings.contains_key(type_name) {
        return Ok(());
    }

    // TODO: Ensure that the type's representation is indeed object
    let type_mapping = data_specification::TypeMapping::Object {
        field_mappings: HashMap::new(),
    };
    type_mappings.insert(
        type_name.clone(),
        resolve_type_mapping(type_name, &type_mapping, types, source_collection_type)?,
    );

    // TODO: Recursively add implicit type mappings for nested object types
    Ok(())
}

fn resolve_model_source(
    model_source: &open_dds::data_specification::ModelSource,
    model: &mut Model,
    data_sources: &HashMap<DataSourceName, DataSourceContext>,
    types: &HashMap<TypeName, TypeRepresentation>,
) -> Result<(), Error> {
    if model.source.is_some() {
        return Err(Error::DuplicateModelSourceDefinition {
            model_name: model.name.clone(),
        });
    }
    let data_source_context = data_sources
        .get(&model_source.data_source_name)
        .ok_or_else(|| Error::UnknownModelDataSource {
            model_name: model.name.clone(),
            data_source: model_source.data_source_name.clone(),
        })?;

    let source_collection = data_source_context
        .data_source
        .schema
        .collections
        .iter()
        .find(|collection_info| collection_info.name == *model_source.collection)
        .ok_or_else(|| Error::UnknownModelCollection {
            model_name: model.name.clone(),
            data_source: model_source.data_source_name.clone(),
            collection: model_source.collection.clone(),
        })?;

    let source_collection_type = data_source_context
        .data_source
        .schema
        .object_types
        .get(&source_collection.collection_type)
        .ok_or(ndc_validation::NDCValidationError::NoSuchType(
            source_collection.collection_type.clone(),
        ))?;

    let mut type_mappings = HashMap::new();
    for (type_name, type_mapping) in &model_source.type_mappings {
        let resolved_type_mapping =
            resolve_type_mapping(type_name, type_mapping, types, source_collection_type).map_err(
                |type_validation_error| Error::ModelTypeMappingValidationError {
                    model_name: model.name.clone(),
                    error: type_validation_error,
                },
            )?;
        if type_mappings
            .insert(type_name.clone(), resolved_type_mapping)
            .is_some()
        {
            return Err(Error::DuplicateTypeMappingDefinition {
                type_name: type_name.clone(),
                model_name: model.name.clone(),
            });
        }
    }

    // Ensure that any unmapped types get implicitly mapped
    add_implicit_type_mappings(
        &model.data_type,
        types,
        &mut type_mappings,
        source_collection_type,
    )
    .map_err(
        |type_validation_error| Error::ModelTypeMappingValidationError {
            model_name: model.name.clone(),
            error: type_validation_error,
        },
    )?;

    let mut argument_mappings = HashMap::new();
    for argument_mapping in &model_source.argument_mappings {
        // Ensure that the argument being mapped actually exists
        if !model.arguments.contains_key(argument_mapping.0) {
            return Err(Error::UnknownModelSourceArgument {
                model_name: model.name.clone(),
                argument_name: argument_mapping.0.clone(),
            });
        }

        if argument_mappings
            .insert(argument_mapping.0.clone(), argument_mapping.1.clone())
            .is_some()
        {
            return Err(Error::DuplicateModelArgumentMapping {
                model_name: model.name.clone(),
                argument_name: argument_mapping.0.clone(),
            });
        }
    }

    for argument_name in model.arguments.keys() {
        // If no mapping is defined for an argument, implicitly assume it maps with
        // the same name
        if !argument_mappings.contains_key(argument_name) {
            argument_mappings.insert(argument_name.clone(), argument_name.0.clone());
        }
    }

    model.source = Some(ModelSource {
        data_source: DataSource {
            name: data_source_context.data_source.name.clone(),
            data_connector_url: data_source_context.data_source.data_connector_url.clone(),
        },
        collection: model_source.collection.clone(),
        type_mappings,
        argument_mappings,
    });

    ndc_validation::validate_ndc(&model.name, model, &data_source_context.data_source.schema)?;
    Ok(())
}

fn resolve_model_graphql_api(
    model_graphql_definition: &data_specification::ModelGraphQlDefinition,
    model: &mut Model,
    existing_graphql_types: &mut HashSet<GraphQlTypeName>,
    data_sources: &HashMap<DataSourceName, DataSourceContext>,
) -> Result<(), Error> {
    let model_name = &model.name;
    if let TypeRepresentation::Object { fields, .. } = &model.type_representation {
        for select_unique in &model_graphql_definition.select_uniques {
            let mut unique_identifier_fields = IndexMap::new();
            for field_name in &select_unique.unique_identifier {
                let field_type = &fields
                    .get(field_name)
                    .ok_or_else(|| Error::UnknownFieldInUniqueIdentifier {
                        model_name: model_name.clone(),
                        field_name: field_name.clone(),
                    })?
                    .field_type;
                if unique_identifier_fields
                    .insert(field_name.clone(), field_type.clone())
                    .is_some()
                {
                    return Err(Error::DuplicateFieldInUniqueIdentifier {
                        model_name: model_name.clone(),
                        field_name: field_name.clone(),
                    });
                }
            }
            model
                .graphql_api
                .select_uniques
                .push(SelectUniqueGraphQlDefinition {
                    query_root_field: select_unique.query_root_field.clone(),
                    unique_identifier: unique_identifier_fields,
                });
        }

        model.graphql_api.order_by_expression = model
            .source
            .as_ref()
            .map(
                |model_source: &ModelSource| -> Result<ModelOrderByExpression, Error> {
                    check_conflicting_graphql_types(
                        existing_graphql_types,
                        model_graphql_definition.order_by_expression_type.as_ref(),
                    )?;
                    let order_by_type_name = model_graphql_definition
                        .order_by_expression_type
                        .clone()
                        .unwrap_or_else(|| {
                            GraphQlTypeName(format!(
                                "{}_order_by", // Default to ModelName_order_by
                                model.name
                            ))
                        });

                    let TypeMapping::Object { field_mappings } = model_source
                        .type_mappings
                        .get(&model.data_type)
                        .ok_or(Error::TypeMappingRequired {
                            model_name: model_name.clone(),
                            type_name: model.data_type.clone(),
                            data_source: model_source.data_source.name.clone(),
                        })?;

                    let mut order_by_fields = HashMap::new();
                    for (field_name, field_mapping) in field_mappings.iter() {
                        order_by_fields.insert(
                            field_name.clone(),
                            OrderByExpressionInfo {
                                ndc_column: field_mapping.column.clone(),
                            },
                        );
                    }

                    Ok(ModelOrderByExpression {
                        data_source_name: model_source.data_source.name.clone(),
                        order_by_type_name,
                        order_by_fields,
                    })
                },
            )
            .transpose()?;

        // record filter expression info
        model.graphql_api.filter_expression = model
            .source
            .as_ref()
            .map(
                |model_source: &ModelSource| -> Result<ModelFilterExpression, Error> {
                    check_conflicting_graphql_types(
                        existing_graphql_types,
                        model_graphql_definition.filter_expression_type.as_ref(),
                    )?;
                    let where_type_name = model_graphql_definition
                        .filter_expression_type
                        .clone()
                        .unwrap_or_else(|| {
                            GraphQlTypeName(format!(
                                "{}_bool_exp", // Default to ModelName_bool_exp
                                model.name
                            ))
                        });
                    let mut scalar_fields = HashMap::new();
                    let scalar_types = &data_sources
                        .get(&model_source.data_source.name)
                        .ok_or(Error::UnknownModelDataSource {
                            model_name: model_name.clone(),
                            data_source: model_source.data_source.name.clone(),
                        })?
                        .scalars;

                    let TypeMapping::Object { field_mappings } = model_source
                        .type_mappings
                        .get(&model.data_type)
                        .ok_or(Error::TypeMappingRequired {
                            model_name: model_name.clone(),
                            type_name: model.data_type.clone(),
                            data_source: model_source.data_source.name.clone(),
                        })?;

                    for (field_name, field_mapping) in field_mappings.iter() {
                        // Generate comparison expression for fields mapped to simple scalar type
                        if let Some(scalar_type_name) = field_mapping
                            .column_type
                            .clone()
                            .and_then(get_simple_scalar)
                        {
                            let scalar_type_info = scalar_types.get(&scalar_type_name).ok_or(
                                Error::UnknownScalarTypeInDataSource {
                                    scalar_type: scalar_type_name.clone(),
                                    data_source: model_source.data_source.name.clone(),
                                },
                            )?;

                            let graphql_type_name = &scalar_type_info
                                .comparison_expression_name
                                .clone()
                                .unwrap_or_else(|| {
                                    GraphQlTypeName(format!(
                                        "{}_{}_comparison_exp",
                                        model_source.data_source.name, scalar_type_name,
                                    ))
                                });

                            let mut operators = BTreeMap::new();
                            for (op_name, op_definition) in
                                scalar_type_info.scalar_type.comparison_operators.iter()
                            {
                                operators.insert(
                                    op_name.clone(),
                                    resolve_ndc_type(
                                        &model_source.data_source.name,
                                        op_definition.argument_type.clone(),
                                        scalar_types,
                                    )?,
                                );
                            }
                            // equal operator
                            operators.insert(
                                "_eq".to_string(),
                                scalar_type_info
                                    .representation
                                    .as_ref()
                                    .ok_or(Error::DataConnectorScalarRepresentationRequired {
                                        data_source: model_source.data_source.name.clone(),
                                        scalar_type: scalar_type_name.clone(),
                                    })?
                                    .clone(),
                            );
                            // is_null operator
                            operators.insert(
                                "_is_null".to_string(),
                                Type::Inbuilt(InbuiltType::Boolean),
                            );

                            // Register scalar comparison field only if it contains non-zero operators.
                            if !operators.is_empty() {
                                scalar_fields.insert(
                                    field_name.clone(),
                                    ComparisonExpressionInfo {
                                        data_source_name: model_source.data_source.name.clone(),
                                        scalar_type_name: scalar_type_name.clone(),
                                        type_name: graphql_type_name.clone(),
                                        ndc_column: field_mapping.column.clone(),
                                        operators,
                                    },
                                );
                            }
                        }
                    }
                    Ok(ModelFilterExpression {
                        where_type_name,
                        scalar_fields,
                    })
                },
            )
            .transpose()?;

        // record select_many root field
        model.graphql_api.select_many =
            model_graphql_definition
                .select_many
                .as_ref()
                .map(|sm| SelectManyGraphQlDefinition {
                    query_root_field: sm.query_root_field.clone(),
                })
    } else {
        return Err(Error::InvalidTypeRepresentation {
            model_name: model_name.clone(),
            type_representation: model.type_representation.clone(),
        });
    }
    if model.arguments.is_empty() {
        if model_graphql_definition.arguments_input_field.is_some()
            || model_graphql_definition.arguments_input_type.is_some()
        {
            return Err(Error::UnnecessaryModelArgumentsGraphQlInputConfiguration {
                model_name: model_name.clone(),
            });
        }
    } else {
        check_conflicting_graphql_types(
            existing_graphql_types,
            model_graphql_definition.arguments_input_type.as_ref(),
        )?;
        model.graphql_api.arguments_input = Some(ModelArgumentsGraphQlInput {
            field: model_graphql_definition
                .arguments_input_field
                .clone()
                .unwrap_or_else(|| GraphQlFieldName("args".to_string())),
            graphql_type: model_graphql_definition
                .arguments_input_type
                .clone()
                .unwrap_or_else(|| GraphQlTypeName(format!("{model_name}_arguments"))),
        });
    }

    Ok(())
}

fn resolve_output_type_permission(
    type_representation: &mut TypeRepresentation,
    output_permission: &data_specification::TypeOutputPermissions,
) -> Result<(), Error> {
    match type_representation {
        TypeRepresentation::ScalarType { .. } => Err(Error::UnsupportedTypeInOutputPermissions {
            type_name: output_permission.type_name.clone(),
        }),
        TypeRepresentation::Object {
            fields,
            type_permissions,
            ..
        } => {
            // validate all the fields definied in output permissions actually
            // exist in this type definition
            for (role, permission) in &output_permission.permissions {
                for field_name in permission.fields.iter() {
                    if !fields.contains_key(field_name) {
                        return Err(Error::UnknownFieldInOutputPermissionsDefinition {
                            field_name: field_name.clone(),
                            type_name: output_permission.type_name.clone(),
                        });
                    }
                }
                if type_permissions
                    .insert(role.clone(), permission.clone())
                    .is_some()
                {
                    return Err(Error::DuplicateOutputTypePermissions {
                        type_name: output_permission.type_name.clone(),
                    });
                }
            }
            Ok(())
        }
    }
}

// helper function to determine whether a ndc type is a simple scalar
fn get_simple_scalar(t: ndc::models::Type) -> Option<String> {
    match t {
        ndc::models::Type::Named { name } => Some(name),
        ndc::models::Type::Nullable { underlying_type } => get_simple_scalar(*underlying_type),
        ndc::models::Type::Array { element_type: _ } => None,
    }
}

// helper function to resolve ndc types to dds type based on scalar type representations
fn resolve_ndc_type(
    data_source: &DataSourceName,
    source_type: ndc::models::Type,
    scalars: &HashMap<String, ScalarTypeInfo>,
) -> Result<Type, Error> {
    match source_type {
        ndc::models::Type::Named { name } => {
            let scalar_type = scalars
                .get(&name)
                .ok_or(Error::UnknownScalarTypeInDataSource {
                    data_source: data_source.clone(),
                    scalar_type: name.clone(),
                })?;

            scalar_type.representation.clone().ok_or(
                Error::DataConnectorScalarRepresentationRequired {
                    data_source: data_source.clone(),
                    scalar_type: name.clone(),
                },
            )
        }
        ndc::models::Type::Nullable { underlying_type } => {
            resolve_ndc_type(data_source, *underlying_type, scalars)
                .map(|ty| Type::Inbuilt(InbuiltType::Nullable(Box::new(ty))))
        }
        ndc::models::Type::Array { element_type } => {
            resolve_ndc_type(data_source, *element_type, scalars)
                .map(|ty| Type::Inbuilt(InbuiltType::Array(Box::new(ty))))
        }
    }
}