# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

require 'avro/logical_types'

module Avro
  class Schema
    # Sets of strings, for backwards compatibility. See below for sets of symbols,
    # for better performance.
    PRIMITIVE_TYPES = Set.new(%w[null boolean string bytes int long float double])
    NAMED_TYPES =     Set.new(%w[fixed enum record error])

    VALID_TYPES = PRIMITIVE_TYPES + NAMED_TYPES + Set.new(%w[array map union request])

    PRIMITIVE_TYPES_SYM = Set.new(PRIMITIVE_TYPES.map(&:to_sym))
    NAMED_TYPES_SYM     = Set.new(NAMED_TYPES.map(&:to_sym))
    VALID_TYPES_SYM     = Set.new(VALID_TYPES.map(&:to_sym))

    NAME_REGEX = /^([A-Za-z_][A-Za-z0-9_]*)(\.([A-Za-z_][A-Za-z0-9_]*))*$/

    INT_MIN_VALUE = -(1 << 31)
    INT_MAX_VALUE = (1 << 31) - 1
    LONG_MIN_VALUE = -(1 << 63)
    LONG_MAX_VALUE = (1 << 63) - 1

    def self.parse(json_string)
      real_parse(MultiJson.load(json_string), {})
    end

    # Build Avro Schema from data parsed out of JSON string.
    def self.real_parse(json_obj, names=nil, default_namespace=nil)
      if json_obj.is_a? Hash
        type = json_obj['type']
        logical_type = json_obj['logicalType']
        raise SchemaParseError, %Q(No "type" property: #{json_obj}) if type.nil?

        # Check that the type is valid before calling #to_sym, since symbols are never garbage
        # collected (important to avoid DoS if we're accepting schemas from untrusted clients)
        unless VALID_TYPES.include?(type)
          raise SchemaParseError, "Unknown type: #{type}"
        end

        type_sym = type.to_sym
        if PRIMITIVE_TYPES_SYM.include?(type_sym)
          case type_sym
          when :bytes
            precision = json_obj['precision']
            scale = json_obj['scale']
            return BytesSchema.new(type_sym, logical_type, precision, scale)
          else
            return PrimitiveSchema.new(type_sym, logical_type)
          end
        elsif NAMED_TYPES_SYM.include? type_sym
          name = json_obj['name']
          if !Avro.disable_schema_name_validation && name !~ NAME_REGEX
            raise SchemaParseError, "Name #{name} is invalid for type #{type}!"
          end
          namespace = json_obj.include?('namespace') ? json_obj['namespace'] : default_namespace
          aliases = json_obj['aliases']
          case type_sym
          when :fixed
            size = json_obj['size']
            return FixedSchema.new(name, namespace, size, names, logical_type, aliases)
          when :enum
            symbols = json_obj['symbols']
            doc     = json_obj['doc']
            default = json_obj['default']
            return EnumSchema.new(name, namespace, symbols, names, doc, default, aliases)
          when :record, :error
            fields = json_obj['fields']
            doc    = json_obj['doc']
            return RecordSchema.new(name, namespace, fields, names, type_sym, doc, aliases)
          else
            raise SchemaParseError.new("Unknown named type: #{type}")
          end

        else
          case type_sym
          when :array
            return ArraySchema.new(json_obj['items'], names, default_namespace)
          when :map
            return MapSchema.new(json_obj['values'], names, default_namespace)
          else
            raise SchemaParseError.new("Unknown Valid Type: #{type}")
          end
        end

      elsif json_obj.is_a? Array
        # JSON array (union)
        return UnionSchema.new(json_obj, names, default_namespace)
      elsif PRIMITIVE_TYPES.include? json_obj
        return PrimitiveSchema.new(json_obj)
      else
        raise UnknownSchemaError.new(json_obj)
      end
    end

    # Determine if a ruby datum is an instance of a schema
    def self.validate(expected_schema, logical_datum, options = { recursive: true, encoded: false })
      SchemaValidator.validate!(expected_schema, logical_datum, options)
      true
    rescue SchemaValidator::ValidationError
      false
    end

    def initialize(type, logical_type=nil)
      @type_sym = type.is_a?(Symbol) ? type : type.to_sym
      @logical_type = logical_type
    end

    attr_reader :type_sym
    attr_reader :logical_type

    # Returns the type as a string (rather than a symbol), for backwards compatibility.
    # Deprecated in favor of {#type_sym}.
    def type; @type_sym.to_s; end

    def type_adapter
      @type_adapter ||= LogicalTypes.type_adapter(type, logical_type) || LogicalTypes::Identity
    end

    # Returns the MD5 fingerprint of the schema as an Integer.
    def md5_fingerprint
      parsing_form = SchemaNormalization.to_parsing_form(self)
      Digest::MD5.hexdigest(parsing_form).to_i(16)
    end

    # Returns the SHA-256 fingerprint of the schema as an Integer.
    def sha256_fingerprint
      parsing_form = SchemaNormalization.to_parsing_form(self)
      Digest::SHA256.hexdigest(parsing_form).to_i(16)
    end

    CRC_EMPTY = 0xc15d213aa4d7a795

    # The java library caches this value after initialized, so this pattern
    # mimics that.
    @@fp_table = nil
    def initFPTable
      @@fp_table = Array.new(256)
      256.times do |i|
        fp = i
        8.times do
          fp = (fp >> 1) ^ ( CRC_EMPTY & -( fp & 1 ) )
        end
        @@fp_table[i] = fp
      end
    end

    def crc_64_avro_fingerprint
      parsing_form = Avro::SchemaNormalization.to_parsing_form(self)
      data_bytes = parsing_form.unpack("C*")

      initFPTable unless @@fp_table

      fp = CRC_EMPTY
      data_bytes.each do |b|
        fp = (fp >> 8) ^ @@fp_table[ (fp ^ b) & 0xff ]
      end
      fp
    end

    SINGLE_OBJECT_MAGIC_NUMBER = [0xC3, 0x01]
    def single_object_encoding_header
      [SINGLE_OBJECT_MAGIC_NUMBER, single_object_schema_fingerprint].flatten
    end
    def single_object_schema_fingerprint
      working = crc_64_avro_fingerprint
      bytes = Array.new(8)
      8.times do |i|
        bytes[i] = (working & 0xff)
        working = working >> 8
      end
      bytes
    end

    def read?(writers_schema)
      SchemaCompatibility.can_read?(writers_schema, self)
    end

    def be_read?(other_schema)
      other_schema.read?(self)
    end

    def mutual_read?(other_schema)
      SchemaCompatibility.mutual_read?(other_schema, self)
    end

    def ==(other, _seen=nil)
      other.is_a?(Schema) && type_sym == other.type_sym
    end

    def hash(_seen=nil)
      type_sym.hash
    end

    def subparse(json_obj, names=nil, namespace=nil)
      if json_obj.is_a?(String) && names
        fullname = Name.make_fullname(json_obj, namespace)
        return names[fullname] if names.include?(fullname)
      end

      begin
        Schema.real_parse(json_obj, names, namespace)
      rescue => e
        raise e if e.is_a? SchemaParseError
        raise SchemaParseError, "Sub-schema for #{self.class.name} not a valid Avro schema. Bad schema: #{json_obj}"
      end
    end

    def to_avro(_names=nil)
      props = {'type' => type}
      props['logicalType'] = logical_type if logical_type
      props
    end

    def to_s
      MultiJson.dump to_avro
    end

    def validate_aliases!
      unless aliases.nil? ||
        (aliases.is_a?(Array) && aliases.all? { |a| a.is_a?(String) })

        raise Avro::SchemaParseError,
              "Invalid aliases value #{aliases.inspect} for #{type} #{name}. Must be an array of strings."
      end
    end
    private :validate_aliases!

    class NamedSchema < Schema
      attr_reader :name, :namespace, :aliases

      def initialize(type, name, namespace=nil, names=nil, doc=nil, logical_type=nil, aliases=nil)
        super(type, logical_type)
        @name, @namespace = Name.extract_namespace(name, namespace)
        @doc = doc
        @aliases = aliases
        validate_aliases! if aliases
        Name.add_name(names, self)
      end

      def to_avro(names=Set.new)
        if @name
          return fullname if names.include?(fullname)
          names << fullname
        end
        props = {'name' => @name}
        props.merge!('namespace' => @namespace) if @namespace
        props['namespace'] = @namespace if @namespace
        props['doc'] = @doc if @doc
        props['aliases'] = aliases if aliases && aliases.any?
        super.merge props
      end

      def fullname
        @fullname ||= Name.make_fullname(@name, @namespace)
      end

      def fullname_aliases
        @fullname_aliases ||= if aliases
                                aliases.map { |a| Name.make_fullname(a, namespace) }
                              else
                                []
                              end
      end

      def match_fullname?(name)
        name == fullname || fullname_aliases.include?(name)
      end
    end

    class RecordSchema < NamedSchema
      attr_reader :fields, :doc

      def self.make_field_objects(field_data, names, namespace=nil)
        field_objects, field_names, alias_names = [], Set.new, Set.new
        field_data.each do |field|
          if field.respond_to?(:[]) # TODO(jmhodges) wtffffff
            type = field['type']
            name = field['name']
            default = field.key?('default') ? field['default'] : :no_default
            order = field['order']
            doc = field['doc']
            aliases = field['aliases']
            new_field = Field.new(type, name, default, order, names, namespace, doc, aliases)
            # make sure field name has not been used yet
            if field_names.include?(new_field.name)
              raise SchemaParseError, "Field name #{new_field.name.inspect} is already in use"
            end
            field_names << new_field.name
            # make sure alias has not be been used yet
            if new_field.aliases && alias_names.intersect?(new_field.aliases.to_set)
              raise SchemaParseError, "Alias #{(alias_names & new_field.aliases).to_a} already in use"
            end
            alias_names.merge(new_field.aliases) if new_field.aliases
          else
            raise SchemaParseError, "Not a valid field: #{field}"
          end
          field_objects << new_field
        end
        field_objects
      end

      def initialize(name, namespace, fields, names=nil, schema_type=:record, doc=nil, aliases=nil)
        if schema_type == :request || schema_type == 'request'
          @type_sym = schema_type.to_sym
          @namespace = namespace
          @name = nil
          @doc = nil
        else
          super(schema_type, name, namespace, names, doc, nil, aliases)
        end
        @fields = if fields
                    RecordSchema.make_field_objects(fields, names, self.namespace)
                  else
                    {}
                  end
      end

      def fields_hash
        @fields_hash ||= fields.inject({}){|hsh, field| hsh[field.name] = field; hsh }
      end

      def fields_by_alias
        @fields_by_alias ||= fields.each_with_object({}) do |field, hash|
          if field.aliases
            field.aliases.each do |a|
              hash[a] = field
            end
          end
        end
      end

      def to_avro(names=Set.new)
        hsh = super
        return hsh unless hsh.is_a?(Hash)
        hsh['fields'] = @fields.map {|f| f.to_avro(names) }
        if type_sym == :request
          hsh['fields']
        else
          hsh
        end
      end
    end

    class ArraySchema < Schema
      attr_reader :items

      def initialize(items, names=nil, default_namespace=nil)
        super(:array)
        @items = subparse(items, names, default_namespace)
      end

      def to_avro(names=Set.new)
        super.merge('items' => items.to_avro(names))
      end
    end

    class MapSchema < Schema
      attr_reader :values

      def initialize(values, names=nil, default_namespace=nil)
        super(:map)
        @values = subparse(values, names, default_namespace)
      end

      def to_avro(names=Set.new)
        super.merge('values' => values.to_avro(names))
      end
    end

    class UnionSchema < Schema
      attr_reader :schemas

      def initialize(schemas, names=nil, default_namespace=nil)
        super(:union)

        @schemas = schemas.each_with_object([]) do |schema, schema_objects|
          new_schema = subparse(schema, names, default_namespace)
          ns_type = new_schema.type_sym

          if VALID_TYPES_SYM.include?(ns_type) &&
              !NAMED_TYPES_SYM.include?(ns_type) &&
              schema_objects.any?{|o| o.type_sym == ns_type }
            raise SchemaParseError, "#{ns_type} is already in Union"
          elsif ns_type == :union
            raise SchemaParseError, "Unions cannot contain other unions"
          else
            schema_objects << new_schema
          end
        end
      end

      def to_avro(names=Set.new)
        schemas.map {|schema| schema.to_avro(names) }
      end
    end

    class EnumSchema < NamedSchema
      SYMBOL_REGEX = /^[A-Za-z_][A-Za-z0-9_]*$/

      attr_reader :symbols, :doc, :default

      def initialize(name, space, symbols, names=nil, doc=nil, default=nil, aliases=nil)
        if symbols.uniq.length < symbols.length
          fail_msg = "Duplicate symbol: #{symbols}"
          raise Avro::SchemaParseError, fail_msg
        end

        if !Avro.disable_enum_symbol_validation
          invalid_symbols = symbols.select { |symbol| symbol !~ SYMBOL_REGEX }

          if invalid_symbols.any?
            raise SchemaParseError,
              "Invalid symbols for #{name}: #{invalid_symbols.join(', ')} don't match #{SYMBOL_REGEX.inspect}"
          end
        end

        if default && !symbols.include?(default)
          raise Avro::SchemaParseError, "Default '#{default}' is not a valid symbol for enum #{name}"
        end

        super(:enum, name, space, names, doc, nil, aliases)
        @default = default
        @symbols = symbols
      end

      def to_avro(_names=Set.new)
        avro = super
        if avro.is_a?(Hash)
          avro['symbols'] = symbols
          avro['default'] = default if default
        end
        avro
      end
    end

    # Valid primitive types are in PRIMITIVE_TYPES.
    class PrimitiveSchema < Schema
      def initialize(type, logical_type=nil)
        if PRIMITIVE_TYPES_SYM.include?(type)
          super(type, logical_type)
        elsif PRIMITIVE_TYPES.include?(type)
          super(type.to_sym, logical_type)
        else
          raise AvroError.new("#{type} is not a valid primitive type.")
        end
      end

      def to_avro(names=nil)
        hsh = super
        hsh.size == 1 ? type : hsh
      end
    end

    class BytesSchema < PrimitiveSchema
      attr_reader :precision, :scale
      def initialize(type, logical_type=nil, precision=nil, scale=nil)
        super(type.to_sym, logical_type)
        @precision = precision
        @scale = scale
      end

      def to_avro(names=nil)
        avro = super
        return avro if avro.is_a?(String)

        avro['precision'] = precision if precision
        avro['scale'] = scale if scale
        avro
      end
    end

    class FixedSchema < NamedSchema
      attr_reader :size
      def initialize(name, space, size, names=nil, logical_type=nil, aliases=nil)
        # Ensure valid cto args
        unless size.is_a?(Integer)
          raise AvroError, 'Fixed Schema requires a valid integer for size property.'
        end
        super(:fixed, name, space, names, nil, logical_type, aliases)
        @size = size
      end

      def to_avro(names=Set.new)
        avro = super
        avro.is_a?(Hash) ? avro.merge('size' => size) : avro
      end
    end

    class Field < Schema
      attr_reader :type, :name, :default, :order, :doc, :aliases

      def initialize(type, name, default=:no_default, order=nil, names=nil, namespace=nil, doc=nil, aliases=nil)
        @type = subparse(type, names, namespace)
        @name = name
        @default = default
        @order = order
        @doc = doc
        @aliases = aliases
        validate_aliases! if aliases
        validate_default! if default? && !Avro.disable_field_default_validation
      end

      def default?
        @default != :no_default
      end

      def to_avro(names=Set.new)
        {'name' => name, 'type' => type.to_avro(names)}.tap do |avro|
          avro['default'] = default if default?
          avro['order'] = order if order
          avro['doc'] = doc if doc
        end
      end

      def alias_names
        @alias_names ||= Array(aliases)
      end

      private

      def validate_default!
        type_for_default = if type.type_sym == :union
                             type.schemas.first
                           else
                             type
                           end

        Avro::SchemaValidator.validate!(type_for_default, default)
      rescue Avro::SchemaValidator::ValidationError => e
        raise Avro::SchemaParseError, "Error validating default for #{name}: #{e.message}"
      end
    end
  end

  class SchemaParseError < AvroError; end

  class UnknownSchemaError < SchemaParseError
    attr_reader :type_name

    def initialize(type)
      @type_name = type
      super("#{type.inspect} is not a schema we know about.")
    end
  end

  module Name
    def self.extract_namespace(name, namespace)
      parts = name.split('.')
      if parts.size > 1
        namespace, name = parts[0..-2].join('.'), parts.last
      end
      return name, namespace
    end

    # Add a new schema object to the names dictionary (in place).
    def self.add_name(names, new_schema)
      new_fullname = new_schema.fullname
      if Avro::Schema::VALID_TYPES.include?(new_fullname)
        raise SchemaParseError, "#{new_fullname} is a reserved type name."
      elsif names.nil?
        names = {}
      elsif names.has_key?(new_fullname)
        raise SchemaParseError, "The name \"#{new_fullname}\" is already in use."
      end

      names[new_fullname] = new_schema
      names
    end

    def self.make_fullname(name, namespace)
      if !name.include?('.') && !namespace.nil?
        namespace + '.' + name
      else
        name
      end
    end
  end
end