# frozen_string_literal: true

module Deimos
  # Represents an object which needs to inform Kafka when it is saved or
  # bulk imported.
  module KafkaSource
    extend ActiveSupport::Concern

    included do
      after_create(:send_kafka_event_on_create)
      after_update(:send_kafka_event_on_update)
      after_destroy(:send_kafka_event_on_destroy)
    end

    # Send the newly created model to Kafka.
    # @return [void]
    def send_kafka_event_on_create
      return unless self.persisted?
      return unless self.class.kafka_config[:create]

      self.truncate_columns if Deimos.config.producers.truncate_columns
      self.class.kafka_producers.each { |p| p.send_event(self) }
    end

    # Send the newly updated model to Kafka.
    # @return [void]
    def send_kafka_event_on_update
      return unless self.class.kafka_config[:update]

      self.truncate_columns if Deimos.config.producers.truncate_columns

      producers = self.class.kafka_producers
      producers.each do |producer|
        fields = producer.watched_attributes(self)
        fields -= ['updated_at']
        # Only send an event if a field we care about was changed.
        any_changes = fields.any? do |field|
          field_change = self.previous_changes[field]
          field_change.present? && field_change[0] != field_change[1]
        end
        next unless any_changes

        producer.send_event(self)
      end
    end

    # Send a deletion (null payload) event to Kafka.
    # @return [void]
    def send_kafka_event_on_destroy
      return unless self.class.kafka_config[:delete]

      self.class.kafka_producers.each do |p|
        generated = p.respond_to?(:generate_deletion_payload) ?
                      p.generate_deletion_payload(self) :
                      self.deletion_payload
        p.publish_list([generated])
      end
    end

    # Payload to send after we are destroyed.
    # @return [Hash]
    def deletion_payload
      { payload_key: self[self.class.primary_key] }
    end

    # :nodoc:
    module ClassMethods
      # @return [Hash]
      def kafka_config
        {
          update: true,
          delete: true,
          import: true,
          create: true
        }
      end

      # @return [Array<Deimos::ActiveRecordProducer>] the producers to run.
      def kafka_producers
        raise MissingImplementationError
      end

      # This is an internal method, part of the activerecord_import gem. It's
      # the one that actually does the importing, having already normalized
      # the inputs (arrays, hashes, records etc.)
      # Basically we want to first do the import, then reload the records
      # and send them to Kafka.
      # @!visibility private
      def import_without_validations_or_callbacks(column_names,
                                                  array_of_attributes,
                                                  options = {})
        results = super
        if !self.kafka_config[:import] || array_of_attributes.empty?
          return results
        end

        # This will contain an array of hashes, where each hash is the actual
        # attribute hash that created the object.
        array_of_hashes = []
        array_of_attributes.each do |array|
          array_of_hashes << column_names.zip(array).to_h.with_indifferent_access
        end
        hashes_with_id, hashes_without_id = array_of_hashes.partition { |arr| arr[:id].present? }

        self.kafka_producers.each { |p| p.send_events(hashes_with_id) }

        if hashes_without_id.any?
          if options[:on_duplicate_key_update].present? &&
             options[:on_duplicate_key_update] != [:updated_at]
            unique_columns = column_names.map(&:to_s) -
                             options[:on_duplicate_key_update].map(&:to_s) - %w(id created_at)
            records = hashes_without_id.map do |hash|
              self.where(unique_columns.map { |c| [c, hash[c]] }.to_h).first
            end
            self.kafka_producers.each { |p| p.send_events(records) }
          else
            # re-fill IDs based on what was just entered into the DB.
            last_id = if self.connection.adapter_name.downcase =~ /sqlite/
                        self.connection.select_value('select last_insert_rowid()') -
                          hashes_without_id.size + 1
                      else
                        # mysql
                        self.connection.select_value('select LAST_INSERT_ID()')
                      end
            hashes_without_id.each_with_index do |attrs, i|
              attrs[:id] = last_id + i
            end
            self.kafka_producers.each { |p| p.send_events(hashes_without_id) }
          end
        end
        results
      end
    end

    # check if any field has value longer than the field limit
    def truncate_columns
      self.class.columns.each do |col|
        next unless col.type == :string
        next if self[col.name].blank?

        if self[col.name].to_s.length > col.limit
          self[col.name] = self[col.name][0..col.limit - 1]
        end
      end
      false
    end
  end
end
