require 'manticore'

module LogStash module PluginMixins module Kafka
  module AvroSchemaRegistry

    def self.included(base)
      base.extend(self)
      base.setup_schema_registry_config
    end

    def setup_schema_registry_config
      # Option to set key to access Schema Registry.
      config :schema_registry_key, :validate => :string

      # Option to set secret to access Schema Registry.
      config :schema_registry_secret, :validate => :password

      # Option to set the endpoint of the Schema Registry.
      # This option permit the usage of Avro Kafka deserializer which retrieve the schema of the Avro message from an
      # instance of schema registry. If this option has value `value_deserializer_class` nor `topics_pattern` could be valued
      config :schema_registry_url, :validate => :uri

      # Option to set the proxy of the Schema Registry.
      # This option permits to define a proxy to be used to reach the schema registry service instance.
      config :schema_registry_proxy, :validate => :uri

      # Option to skip validating the schema registry during registration. This can be useful when using
      # certificate based auth
      config :schema_registry_validation, :validate => ['auto', 'skip'], :default => 'auto'
    end

    def check_schema_registry_parameters
      if @schema_registry_url
        check_for_schema_registry_conflicts
        @schema_registry_proxy_host, @schema_registry_proxy_port  = split_proxy_into_host_and_port(schema_registry_proxy)
        check_for_key_and_secret
        check_for_schema_registry_connectivity_and_subjects if schema_registry_validation?
      end
    end

    def schema_registry_validation?
      return false if schema_registry_validation.to_s == 'skip'
      return false if using_kerberos? # pre-validation doesn't support kerberos

      true
    end

    def using_kerberos?
      security_protocol == "SASL_PLAINTEXT" || security_protocol == "SASL_SSL"
    end

    private
    def check_for_schema_registry_conflicts
      if @value_deserializer_class != LogStash::Inputs::Kafka::DEFAULT_DESERIALIZER_CLASS
        raise LogStash::ConfigurationError, 'Option schema_registry_url prohibit the customization of value_deserializer_class'
      end
      if @topics_pattern && !@topics_pattern.empty?
       raise LogStash::ConfigurationError, 'Option schema_registry_url prohibit the customization of topics_pattern'
      end
    end

    private
    def check_for_schema_registry_connectivity_and_subjects
      options = {}
      if schema_registry_proxy && !schema_registry_proxy.empty?
        options[:proxy] = schema_registry_proxy.to_s
      end
      if schema_registry_key and !schema_registry_key.empty?
        options[:auth] = {:user => schema_registry_key, :password => schema_registry_secret.value}
      end
      client = Manticore::Client.new(options)
      begin
        response = client.get(@schema_registry_url.uri.to_s + '/subjects').body
      rescue Manticore::ManticoreException => e
        raise LogStash::ConfigurationError.new("Schema registry service doesn't respond, error: #{e.message}")
      end
      registered_subjects = JSON.parse response
      expected_subjects = @topics.map { |t| "#{t}-value"}
      if (expected_subjects & registered_subjects).size != expected_subjects.size
        undefined_topic_subjects = expected_subjects - registered_subjects
        raise LogStash::ConfigurationError, "The schema registry does not contain definitions for required topic subjects: #{undefined_topic_subjects}"
      end
    end

    def split_proxy_into_host_and_port(proxy_uri)
      return nil unless proxy_uri && !proxy_uri.empty?

      port = proxy_uri.port

      host_spec = ""
      host_spec << proxy_uri.scheme || "http"
      host_spec << "://"
      host_spec << "#{proxy_uri.userinfo}@" if proxy_uri.userinfo
      host_spec << proxy_uri.host

      [host_spec, port]
    end

    def check_for_key_and_secret
      if schema_registry_key and !schema_registry_key.empty?
        if !schema_registry_secret or schema_registry_secret.value.empty?
          raise LogStash::ConfigurationError, "Setting `schema_registry_secret` is required when `schema_registry_key` is provided."
        end
      end
    end

  end
end end end