# This file is distributed under New Relic's license terms.
# See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
# frozen_string_literal: true

require 'timeout'
require_relative '../../test_helper'
require 'new_relic/agent/pipe_channel_manager'
require 'new_relic/agent/transaction_event_primitive'

class NewRelic::Agent::PipeChannelManagerTest < Minitest::Test
  include TransactionSampleTestHelper

  def setup
    @test_config = {:monitor_mode => true}
    NewRelic::Agent.agent.drop_buffered_data
    NewRelic::Agent.config.add_config_for_testing(@test_config)
    NewRelic::Agent.config.notify_server_source_added

    listener = NewRelic::Agent::PipeChannelManager::Listener.new
    NewRelic::Agent::PipeChannelManager.instance_variable_set(:@listener, listener)

    NewRelic::Agent.manual_start
    NewRelic::Agent::Tracer.clear_state
  end

  def teardown
    NewRelic::Agent::PipeChannelManager.listener.close_all_pipes
    NewRelic::Agent::PipeChannelManager.listener.stop
    NewRelic::Agent.shutdown
    NewRelic::Agent.config.remove_config(@test_config)
  end

  def test_registering_a_pipe
    NewRelic::Agent::PipeChannelManager.listener.wake.in.expects(:<<).with('.')
    NewRelic::Agent::PipeChannelManager.register_report_channel(1)
    pipe = NewRelic::Agent::PipeChannelManager.channels[1]

    assert_kind_of(IO, pipe.out)
    assert_kind_of(IO, pipe.in)

    NewRelic::Agent::PipeChannelManager.listener.close_all_pipes
  end

  if NewRelic::LanguageSupport.can_fork?
    def test_listener_merges_timeslice_metrics
      metric = 'Custom/test/method'

      NewRelic::Agent.record_metric(metric, 1.0)

      start_listener_with_pipe(666)

      run_child(666) do
        NewRelic::Agent.after_fork
        new_engine = NewRelic::Agent::StatsEngine.new
        new_engine.tl_record_unscoped_metrics(metric, 2.0)
        service = NewRelic::Agent::PipeService.new(666)
        service.metric_data(new_engine.harvest!)
      end

      assert_metrics_recorded(metric => {:total_call_time => 3.0})
    end

    def test_listener_merges_transaction_traces
      sampler = NewRelic::Agent.agent.transaction_sampler

      assert_equal(0, sampler.count)

      start_listener_with_pipe(667)
      run_child(667) do
        NewRelic::Agent.after_fork
        with_config(:'transaction_tracer.transaction_threshold' => 0.0) do
          in_transaction {}
          service = NewRelic::Agent::PipeService.new(667)
          service.transaction_sample_data(sampler.harvest!)
        end
      end

      assert_equal(1, sampler.count)
    end

    def test_listener_merges_error_traces
      skip_for_ruby_2_5_9_and_rails_6_x

      sampler = NewRelic::Agent.agent.error_collector
      sampler.notice_error(Exception.new('message'), :uri => '/myurl/',
        :metric => 'path', :referer => 'test_referer',
        :request_params => {:x => 'y'})
      NewRelic::Agent.agent.merge_data_for_endpoint(:error_data, sampler.error_trace_aggregator.harvest!)

      errors = NewRelic::Agent.agent.error_collector.error_trace_aggregator.instance_variable_get(:@errors)

      assert_equal(1, errors.size)

      start_listener_with_pipe(668)

      run_child(668) do
        NewRelic::Agent.after_fork
        new_sampler = NewRelic::Agent::ErrorCollector.new(NewRelic::Agent.instance.events)
        new_sampler.notice_error(Exception.new('new message'), :uri => '/myurl/',
          :metric => 'path', :referer => 'test_referer',
          :request_params => {:x => 'y'})
        service = NewRelic::Agent::PipeService.new(668)
        service.error_data(new_sampler.error_trace_aggregator.harvest!)
      end
      errors = NewRelic::Agent.agent.error_collector.error_trace_aggregator.instance_variable_get(:@errors)

      assert_equal(2, errors.size, "Expected 2 errors, got #{errors.inspect}")
    end

    def test_listener_merges_analytics_events
      transaction_event_aggregator = NewRelic::Agent.agent.transaction_event_aggregator
      reset_lifetime_counts!(transaction_event_aggregator)

      start_listener_with_pipe(699)
      NewRelic::Agent.agent.stubs(:connected?).returns(true)
      run_child(699) do
        NewRelic::Agent.after_fork(:report_to_channel => 699)
        transaction_event_aggregator.record(event: NewRelic::Agent::TransactionEventPrimitive.create({
          :start_timestamp => Process.clock_gettime(Process::CLOCK_REALTIME),
          :name => 'whatever',
          :duration => 10,
          :type => :controller,
          :priority => 0.5
        }))
        NewRelic::Agent.agent.send(:transmit_analytic_event_data)
      end

      _, samples = transaction_event_aggregator.harvest!

      assert_equal(1, samples.size)
      assert_lifetime_counts(transaction_event_aggregator, 1)
    end

    def test_listener_merges_error_events
      skip_for_ruby_2_5_9_and_rails_6_x

      error_event_aggregator = NewRelic::Agent.agent.error_collector.error_event_aggregator
      reset_lifetime_counts!(error_event_aggregator)

      sampler = NewRelic::Agent.agent.error_collector
      sampler.notice_error(Exception.new('message'), :uri => '/myurl/',
        :metric => 'path', :referer => 'test_referer',
        :request_params => {:x => 'y'})

      start_listener_with_pipe(668)

      run_child(668) do
        NewRelic::Agent.after_fork
        new_sampler = NewRelic::Agent::ErrorCollector.new(NewRelic::Agent.instance.events)
        new_sampler.notice_error(Exception.new('new message'), :uri => '/myurl/',
          :metric => 'path', :referer => 'test_referer',
          :request_params => {:x => 'y'})
        service = NewRelic::Agent::PipeService.new(668)
        service.error_event_data(new_sampler.error_event_aggregator.harvest!)
      end
      _, errors = error_event_aggregator.harvest!

      assert_equal(2, errors.size, "Expected 2 errors, got #{errors.inspect}")
      assert_lifetime_counts(error_event_aggregator, 2)
    end

    def test_listener_merges_sql_traces
      sampler = NewRelic::Agent.agent.sql_sampler
      create_sql_sample(sampler)

      start_listener_with_pipe(667)
      run_child(667) do
        NewRelic::Agent.after_fork
        create_sql_sample(sampler)
        service = NewRelic::Agent::PipeService.new(667)
        service.sql_trace_data(sampler.harvest!)
      end

      assert_equal(2, sampler.harvest!.size)
    end

    def test_close_pipe_on_child_explicit_close
      listener = start_listener_with_pipe(669)
      pid = Process.fork do
        NewRelic::Agent::PipeService.new(669)
      end
      Process.wait(pid)
      listener.stop_listener_thread

      assert_pipe_finished(669)
    end

    def test_close_pipe_on_child_exit
      listener = start_listener_with_pipe(669)
      pid = Process.fork do
        NewRelic::Agent::PipeService.new(669)
        exit!
      end
      Process.wait(pid)
      listener.stop_listener_thread

      assert_pipe_finished(669)
    end

    def test_manager_does_not_crash_when_given_bad_data
      listener = start_listener_with_pipe(670)
      pid = Process.fork do
        listener.pipes[670].in << 'some unloadable garbage'
      end
      Process.wait(pid)
      listener.stop
    end

    def test_manager_does_not_crash_when_given_unmarshallable_junk
      listener = start_listener_with_pipe(671)
      expects_logging(:error, any_parameters)

      pid = Process.fork do
        listener.pipes[671].write("\x00")
      end
      Process.wait(pid)
      listener.stop
    end

    def pipe_finished?(id)
      (!NewRelic::Agent::PipeChannelManager.channels[id] ||
        NewRelic::Agent::PipeChannelManager.channels[id].closed?)
    end

    def assert_pipe_finished(id)
      assert(pipe_finished?(id),
        "Expected pipe with ID #{id} to be nil or closed")
    end

    def create_sql_sample(sampler)
      state = NewRelic::Agent::Tracer.state
      sampler.on_start_transaction(state)
      sampler.notice_sql('SELECT * FROM table', 'ActiveRecord/Widgets/find', nil, 100, state)
      sampler.on_finishing_transaction(state, 'noodles')
    end
  end

  def start_listener_with_pipe(pipe_id)
    listener = NewRelic::Agent::PipeChannelManager.listener
    listener.start
    listener.register_pipe(pipe_id)
    listener
  end

  def test_pipe_read_length_failure
    write_pipe = stub(:set_encoding => nil, :closed? => false, :close => nil)

    # If we only read three bytes, it isn't valid.
    # We can't tell whether any four bytes or more are a "good" length or not.
    read_pipe = stub(:read => 'jrc')
    IO.stubs(:pipe).returns([read_pipe, write_pipe])

    # Includes the failed bytes
    expects_logging(:error, includes('[6a 72 63]'))

    pipe = NewRelic::Agent::PipeChannelManager::Pipe.new

    assert_nil pipe.read
  end

  def test_pipe_read_length_nil_fails
    write_pipe = stub(:set_encoding => nil, :closed? => false, :close => nil)

    # No length at all returned on pipe, also a failure.
    read_pipe = stub(:read => nil)
    IO.stubs(:pipe).returns([read_pipe, write_pipe])

    pipe = NewRelic::Agent::PipeChannelManager::Pipe.new

    assert_nil pipe.read
  end

  def test_listener_pipes_race_condition
    begin
      listener = NewRelic::Agent::PipeChannelManager.listener
      listener.instance_variable_set(:@select_timeout, 0.00001)
      listener.start

      listener.register_pipe(0)
      pipe_out = listener.pipes[0].out

      mutex = Mutex.new
      mutex.lock
      tried_to_close = false
      pipe_out.singleton_class.send(:define_method, :closed?, proc do
        tried_to_close = true
        mutex.synchronize do
        end
      end)

      # Make sure we're actually paused in closed? call
      until tried_to_close
      end

      t = Thread.new do
        listener.register_pipe(1)
      end

      mutex.unlock
      t.join
    ensure
      listener.close_all_pipes
      listener.stop
    end
  end

  def run_child(channel_id)
    pid = Process.fork do
      yield
    end

    Process.wait(pid)
    until pipe_finished?(channel_id)
      sleep(0.01)
    end
  end

  def test_blocking_error_rescued
    desired_error_messages_seen = 0

    wake_mock = MiniTest::Mock.new
    out_mock = MiniTest::Mock.new
    4.times { wake_mock.expect(:out, out_mock) }
    error_stub = proc { |msg| desired_error_messages_seen += 1 if /^(?:Issue while|Ready pipes)/.match?(msg) }
    ready_pipes_mock = MiniTest::Mock.new

    ::IO.stub(:select, [ready_pipes_mock]) do
      ready_pipes_mock.expect(:each, nil)
      ready_pipes_mock.expect(:map, [])
      ready_pipes_mock.expect(:include?, true) do
        true
      end

      def out_mock.read_nonblock(num)
        return unless num == 1

        error = ::Class.new(::StandardError) do
          include ::IO::WaitReadable
        end

        raise error
      end

      NewRelic::Agent.logger.stub(:error, error_stub) do
        listener = NewRelic::Agent::PipeChannelManager::Listener.new
        listener.instance_variable_set(:@wake, wake_mock)
        listener.start
        listener.instance_variable_get(:@thread).join
      end
    end

    assert_equal 2, desired_error_messages_seen
  end

  def assert_lifetime_counts(container, value)
    buffer = container.instance_variable_get(:@buffer)

    assert_equal value, buffer.captured_lifetime
    assert_equal value, buffer.seen_lifetime
  end

  def reset_lifetime_counts!(container)
    buffer = container.instance_variable_get(:@buffer)
    buffer.instance_variable_set(:@captured_lifetime, 0)
    buffer.instance_variable_set(:@seen_lifetime, 0)
  end

  def skip_for_ruby_2_5_9_and_rails_6_x
    # TODO: MAJOR VERSION - remove this method and its calls when Ruby 2.5 is no longer supported
    skip if RUBY_VERSION == '2.5.9' &&
      defined?(Rails::VERSION::STRING) &&
      # version string starts with 6.0 or 6.1
      Rails::VERSION::STRING =~ /\A6\.(0|1)/
  end
end
