# frozen_string_literal: true

module Karafka
  module Instrumentation
    # Default listener that hooks up to our instrumentation and uses its events for logging
    # It can be removed/replaced or anything without any harm to the Karafka app flow.
    class LoggerListener
      # Log levels that we use in this particular listener
      USED_LOG_LEVELS = %i[
        debug
        info
        warn
        error
        fatal
      ].freeze

      private_constant :USED_LOG_LEVELS

      # @param log_polling [Boolean] should we log the fact that messages are being polled. This is
      #   usually noisy and not useful in production but can be useful in dev. While users can
      #   do this themselves this has been requested and asked for often, thus similar to how
      #   extensive logging can be disabled in WaterDrop, we do it here as well.
      def initialize(log_polling: true)
        @log_polling = log_polling
      end

      #
      #
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_connection_listener_before_fetch_loop(event)
        listener_id = event[:caller].id
        subscription_group = event[:subscription_group]
        consumer_group_id = subscription_group.consumer_group.id
        topics = subscription_group.topics.select(&:active?).map(&:name).join(', ')
        group_details = "#{consumer_group_id}/#{subscription_group.id}"

        info(
          "[#{listener_id}] Group #{group_details} subscribing to topics: #{topics}"
        )
      end

      # Logs each messages fetching attempt
      #
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_connection_listener_fetch_loop(event)
        return unless log_polling?

        listener_id = event[:caller].id
        debug "[#{listener_id}] Polling messages..."
      end

      # Logs about messages that we've received from Kafka
      #
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_connection_listener_fetch_loop_received(event)
        return unless log_polling?

        listener_id = event[:caller].id
        time = event[:time].round(2)
        messages_count = event[:messages_buffer].size

        message = "[#{listener_id}] Polled #{messages_count} messages in #{time}ms"

        # We don't want the "polled 0" in dev as it would spam the log
        # Instead we publish only info when there was anything we could poll and fail over to the
        # zero notifications when in debug mode
        messages_count.zero? ? debug(message) : info(message)
      end

      # Prints info about the fact that a given job has started
      #
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_worker_process(event)
        job = event[:job]
        job_type = job.class.to_s.split('::').last
        consumer = job.executor.topic.consumer
        topic = job.executor.topic.name
        partition = job.executor.partition
        info "[#{job.id}] #{job_type} job for #{consumer} on #{topic}-#{partition} started"
      end

      # Prints info about the fact that a given job has finished
      #
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_worker_processed(event)
        job = event[:job]
        time = event[:time].round(2)
        job_type = job.class.to_s.split('::').last
        consumer = job.executor.topic.consumer
        topic = job.executor.topic.name
        partition = job.executor.partition
        info <<~MSG.tr("\n", ' ').strip!
          [#{job.id}] #{job_type} job for #{consumer}
          on #{topic}-#{partition} finished in #{time} ms
        MSG
      end

      # Prints info about a consumer pause occurrence. Irrelevant if user or system initiated.
      #
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      # @note There may be no offset provided in case user wants to pause on the consecutive offset
      #   position. This can be beneficial when not wanting to purge the buffers.
      def on_client_pause(event)
        topic = event[:topic]
        partition = event[:partition]
        offset = event[:offset]
        client = event[:caller]

        info <<~MSG.tr("\n", ' ').strip!
          [#{client.id}]
          Pausing on topic #{topic}-#{partition}
          on #{offset ? "offset #{offset}" : 'the consecutive offset'}
        MSG
      end

      # Prints information about resuming of processing of a given topic partition
      #
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_client_resume(event)
        topic = event[:topic]
        partition = event[:partition]
        client = event[:caller]

        info <<~MSG.tr("\n", ' ').strip!
          [#{client.id}] Resuming on topic #{topic}-#{partition}
        MSG
      end

      # Prints info about retry of processing after an error
      #
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_consumer_consuming_retry(event)
        topic = event[:topic]
        partition = event[:partition]
        offset = event[:offset]
        consumer = event[:caller]
        timeout = event[:timeout]

        info <<~MSG.tr("\n", ' ').strip!
          [#{consumer.id}] Retrying of #{consumer.class} after #{timeout} ms
          on topic #{topic}-#{partition} from offset #{offset}
        MSG
      end

      # Prints info about seeking to a particular location
      #
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_consumer_consuming_seek(event)
        topic = event[:topic]
        partition = event[:partition]
        seek_offset = event[:message].offset
        consumer = event[:caller]

        info <<~MSG.tr("\n", ' ').strip!
          [#{consumer.id}] Seeking from #{consumer.class}
          on topic #{topic}-#{partition} to offset #{seek_offset}
        MSG
      end

      # Logs info about system signals that Karafka received and prints backtrace for threads in
      # case of ttin
      #
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_process_notice_signal(event)
        server_id = Karafka::Server.id
        info "[#{server_id}] Received #{event[:signal]} system signal"

        # We print backtrace only for ttin
        return unless event[:signal] == :SIGTTIN

        # Inspired by Sidekiq
        Thread.list.each do |thread|
          tid = (thread.object_id ^ ::Process.pid).to_s(36)

          warn ''
          warn "Thread TID-#{tid} #{thread.name}"

          if thread.backtrace && !thread.backtrace.empty?
            warn thread.backtrace.join("\n")
          else
            warn '<no backtrace available>'
          end
        end
      end

      # Logs info that we're running Karafka app.
      #
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_app_running(event)
        server_id = event[:server_id]

        info "[#{server_id}] Running in #{RUBY_DESCRIPTION}"
        info "[#{server_id}] Running Karafka #{Karafka::VERSION} server"

        return if Karafka.pro?

        info "[#{server_id}] See LICENSE and the LGPL-3.0 for licensing details"
      end

      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_app_quieting(event)
        info "[#{event[:server_id]}] Switching to quiet mode. New messages will not be processed"
      end

      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_app_quiet(event)
        info "[#{event[:server_id]}] Reached quiet mode. No messages will be processed anymore"
      end

      # Logs info that we're going to stop the Karafka server.
      #
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_app_stopping(event)
        info "[#{event[:server_id]}] Stopping Karafka server"
      end

      # Logs info that we stopped the Karafka server.
      #
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_app_stopped(event)
        info "[#{event[:server_id]}] Stopped Karafka server"
      end

      # Logs info about partitions we have lost
      #
      # @param event [Karafka::Core::Monitoring::Event] event details with revoked partitions
      def on_rebalance_partitions_revoked(event)
        revoked_partitions = event[:tpl].to_h.transform_values { |part| part.map(&:partition) }
        group_id = event[:consumer_group_id]
        client_id = event[:client_id]
        group_prefix = "[#{client_id}] Group #{group_id} rebalance"

        if revoked_partitions.empty?
          info "#{group_prefix}: No partitions revoked"
        else
          revoked_partitions.each do |topic, partitions|
            info "#{group_prefix}: #{topic}-[#{partitions.join(',')}] revoked"
          end
        end
      end

      # Logs info about partitions that we've gained
      #
      # @param event [Karafka::Core::Monitoring::Event] event details with assigned partitions
      def on_rebalance_partitions_assigned(event)
        assigned_partitions = event[:tpl].to_h.transform_values { |part| part.map(&:partition) }
        group_id = event[:consumer_group_id]
        client_id = event[:client_id]
        group_prefix = "[#{client_id}] Group #{group_id} rebalance"

        if assigned_partitions.empty?
          info "#{group_prefix}: No partitions assigned"
        else
          assigned_partitions.each do |topic, partitions|
            info "#{group_prefix}: #{topic}-[#{partitions.join(',')}] assigned"
          end
        end
      end

      # Logs info when we have dispatched a message the the DLQ
      #
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_dead_letter_queue_dispatched(event)
        consumer = event[:caller]
        topic = consumer.topic.name
        message = event[:message]
        offset = message.offset
        dlq_topic = consumer.topic.dead_letter_queue.topic
        partition = message.partition

        info <<~MSG.tr("\n", ' ').strip!
          [#{consumer.id}] Dispatched message #{offset}
          from #{topic}-#{partition}
          to DLQ topic: #{dlq_topic}
        MSG
      end

      # Logs info about throttling event
      #
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_filtering_throttled(event)
        consumer = event[:caller]
        topic = consumer.topic.name
        # Here we get last message before throttle
        message = event[:message]
        partition = message.partition
        offset = message.offset

        info <<~MSG.tr("\n", ' ').strip!
          [#{consumer.id}] Throttled and will resume
          from message #{offset}
          on #{topic}-#{partition}
        MSG
      end

      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_filtering_seek(event)
        consumer = event[:caller]
        topic = consumer.topic.name
        # Message to which we seek
        message = event[:message]
        partition = message.partition
        offset = message.offset

        info <<~MSG.tr("\n", ' ').strip!
          [#{consumer.id}] Post-filtering seeking to message #{offset}
          on #{topic}-#{partition}
        MSG
      end

      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_swarm_manager_stopping(event)
        node = event[:node]
        error "Swarm manager detected unhealthy node #{node.pid}. Sending TERM signal..."
      end

      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_swarm_manager_terminating(event)
        node = event[:node]
        error "Swarm manager detected unresponsive node #{node.pid}. Sending KILL signal..."
      end

      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_swarm_manager_before_fork(event)
        debug "Swarm manager starting node with id: #{event[:node].id}"
      end

      # @param _event [Karafka::Core::Monitoring::Event] event details including payload
      def on_swarm_node_after_fork(_event)
        info "Swarm node #{::Process.pid} forked from #{::Process.ppid}"
      end

      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_swarm_manager_control(event)
        pids = event[:caller].nodes.map(&:pid).join(', ')
        debug "Swarm manager checking nodes: #{pids}"
      end

      # There are many types of errors that can occur in many places, but we provide a single
      # handler for all of them to simplify error instrumentation.
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      def on_error_occurred(event)
        type = event[:type]
        error = event[:error]
        backtrace = (error.backtrace || []).join("\n")

        details = [error.to_s, error_details(event)].compact.join(' ')

        case type
        when 'consumer.initialized.error'
          error "Consumer initialized error: #{details}"
          error backtrace
        when 'consumer.wrap.error'
          error "Consumer wrap failed due to an error: #{details}"
          error backtrace
        when 'consumer.consume.error'
          error "Consumer consuming error: #{details}"
          error backtrace
        when 'consumer.revoked.error'
          error "Consumer on revoked failed due to an error: #{details}"
          error backtrace
        when 'consumer.idle.error'
          error "Consumer idle failed due to an error: #{details}"
          error backtrace
        when 'consumer.shutdown.error'
          error "Consumer on shutdown failed due to an error: #{details}"
          error backtrace
        when 'consumer.tick.error'
          error "Consumer on tick failed due to an error: #{details}"
          error backtrace
        when 'consumer.eofed.error'
          error "Consumer on eofed failed due to an error: #{details}"
          error backtrace
        when 'consumer.after_consume.error'
          error "Consumer on after_consume failed due to an error: #{details}"
          error backtrace
        when 'worker.process.error'
          fatal "Worker processing failed due to an error: #{details}"
          fatal backtrace
        when 'connection.listener.fetch_loop.error'
          error "Listener fetch loop error: #{details}"
          error backtrace
        when 'swarm.supervisor.error'
          fatal "Swarm supervisor crashed due to an error: #{details}"
          fatal backtrace
        when 'runner.call.error'
          fatal "Runner crashed due to an error: #{details}"
          fatal backtrace
        when 'app.stopping.error'
          # Counts number of workers and listeners that were still active when forcing the
          # shutdown. Please note, that unless all listeners are closed, workers will not finalize
          # their operations as well.
          # We need to check if listeners and workers are assigned as during super early stages of
          # boot they are not.
          listeners = Server.listeners ? Server.listeners.count(&:active?) : 0
          workers = Server.workers ? Server.workers.count(&:alive?) : 0

          message = <<~MSG.tr("\n", ' ').strip!
            Forceful Karafka server stop with:
            #{workers} active workers and
            #{listeners} active listeners
          MSG

          error message
        when 'app.forceful_stopping.error'
          error "Forceful shutdown error occurred: #{details}"
          error backtrace
        when 'librdkafka.error'
          error "librdkafka internal error occurred: #{details}"
          error backtrace
        # Those can occur when emitted statistics are consumed by the end user and the processing
        # of statistics fails. The statistics are emitted from librdkafka main loop thread and
        # any errors there crash the whole thread
        when 'callbacks.statistics.error'
          error "callbacks.statistics processing failed due to an error: #{details}"
          error backtrace
        when 'callbacks.error.error'
          error "callbacks.error processing failed due to an error: #{details}"
          error backtrace
        # Those will only occur when retries in the client fail and when they did not stop after
        # back-offs
        when 'connection.client.poll.error'
          error "Data polling error occurred: #{details}"
          error backtrace
        when 'connection.client.rebalance_callback.error'
          error "Rebalance callback error occurred: #{details}"
          error backtrace
        when 'connection.client.unsubscribe.error'
          error "Client unsubscribe error occurred: #{details}"
          error backtrace
        when 'parallel_segments.reducer.error'
          error "Parallel segments reducer error occurred: #{details}"
          error backtrace
        when 'parallel_segments.partitioner.error'
          error "Parallel segments partitioner error occurred: #{details}"
          error backtrace
        when 'virtual_partitions.partitioner.error'
          error "Virtual partitions partitioner error occurred: #{details}"
          error backtrace
        # This handles any custom errors coming from places like Web-UI, etc
        else
          error "#{type} error occurred: #{error.class} - #{details}"
          error backtrace
        end
      end

      USED_LOG_LEVELS.each do |log_level|
        define_method log_level do |*args|
          Karafka.logger.send(log_level, *args)
        end
      end

      private

      # @return [Boolean] should we log polling
      def log_polling?
        @log_polling
      end

      # Extracts some structural location
      # @param event [Karafka::Core::Monitoring::Event] event details including payload
      # @note It uses similar approach to the Web UI but here we collect less info because it goes
      #   to the logs.
      def error_details(event)
        caller_ref = event[:caller]

        # Collect extra info if it was a consumer related error.
        # Those come from user code
        details = case caller_ref
                  when Karafka::BaseConsumer
                    extract_consumer_info(caller_ref)
                  when Karafka::Connection::Client
                    extract_client_info(caller_ref)
                  when Karafka::Connection::Listener
                    extract_listener_info(caller_ref)
                  else
                    {}
                  end

        return nil if details.empty?

        "[#{details.map { |label, value| "#{label}: #{value}" }.join(', ')}]"
      end

      # @param consumer [::Karafka::BaseConsumer]
      # @return [Hash] hash with consumer specific info for details of error
      def extract_consumer_info(consumer)
        {
          consumer_group: consumer.topic.consumer_group.id,
          subscription_group: consumer.topic.subscription_group.id,
          topic: consumer.topic.name,
          partition: consumer.partition,
          first_offset: consumer.messages.metadata.first_offset,
          last_offset: consumer.messages.metadata.last_offset
        }
      end

      # @param client [::Karafka::Connection::Client]
      # @return [Hash] hash with client specific info for details of error
      def extract_client_info(client)
        {
          consumer_group: client.subscription_group.consumer_group.id,
          subscription_group: client.subscription_group.id
        }
      end

      # @param listener [::Karafka::Connection::Listener]
      # @return [Hash] hash with listener specific info for details of error
      def extract_listener_info(listener)
        {
          consumer_group: listener.subscription_group.consumer_group.id,
          subscription_group: listener.subscription_group.id
        }
      end
    end
  end
end
