class LittleGhost::Instrumentation::Bus

# File lib/little_ghost/instrumentation.rb, line 93
def initialize(subscribers: [], content_capture: Support::ContentCapture.disabled)
  @subscribers = []
  @content_capture = content_capture
  @handles = {}
  @children = Hash.new(0)
  @finishing = {}
  @mutex = Mutex.new
  @reported_failures = {}
  @shutdown = false
  Array(subscribers).each { |subscriber| subscribe(subscriber) }
end

# File lib/little_ghost/instrumentation.rb, line 226
def active?(handle = nil)
  @mutex.synchronize do
    handle ? @handles[handle.operation_id].equal?(handle) : !@handles.empty?
  end
end

# File lib/little_ghost/instrumentation.rb, line 125
def capture_content(policy)
  raise ArgumentError, "content capture policy must respond to capture" unless policy.respond_to?(:capture)

  @mutex.synchronize { @content_capture = policy }
  policy
end

# File lib/little_ghost/instrumentation.rb, line 215
def context
  deep_copy(ExecutionState[context_key] || {})
end

# File lib/little_ghost/instrumentation.rb, line 220
def current
  ExecutionState[current_key]
end

# File lib/little_ghost/instrumentation.rb, line 181
def finish(handle, diagnostic: nil, **attributes)
  validate_finish!(handle)
  validated = true
  values = handle.payload.merge(attributes).merge(duration_ms: elapsed_ms(handle.started_at))
  values = prepare_attributes(values.compact, diagnostic:)
  notify(:finish, handle.name, values)
  values
ensure
  complete(handle) if validated && active_handle?(handle)
end

# File lib/little_ghost/instrumentation.rb, line 234
def flush(timeout: nil)
  deadline = monotonic_time + Float(timeout) if timeout
  subscribers.each do |subscriber|
    remaining = deadline && [deadline - monotonic_time, 0].max
    notify_subscriber(subscriber, :flush, timeout: remaining)
  end
end

# File lib/little_ghost/instrumentation.rb, line 193
def instrument(name, payload = {})
  values = payload.dup
  handle = start(name, **values)
  result = yield values if block_given?
  handle.finish(**values)
  result
rescue => error
  values ||= payload.dup
  values[:outcome] ||= :error
  values[:error_type] ||= error.class.name
  values[:diagnostic] ||= {exception: diagnostic_exception(error)}
  handle.finish(**values) if handle && active_handle?(handle)
  raise
end

# File lib/little_ghost/instrumentation.rb, line 133
def publish(name, diagnostic: nil, **attributes)
  current_handle = current
  values = context.merge(attributes)
  values[:operation_id] ||= current_handle&.operation_id
  values = prepare_attributes(values.compact, diagnostic:)
  notify(:emit, name.to_sym, values)
  values
end

# File lib/little_ghost/instrumentation.rb, line 243
def shutdown(timeout: nil)
  should_shutdown = @mutex.synchronize do
    return if @shutdown
    raise Error, "cannot shut down instrumentation with active operations" unless @handles.empty?

    @shutdown = true
  end
  return unless should_shutdown

  deadline = monotonic_time + Float(timeout) if timeout
  subscribers.reverse_each do |subscriber|
    remaining = deadline && [deadline - monotonic_time, 0].max
    notify_subscriber(subscriber, :shutdown, timeout: remaining)
  end
end

# File lib/little_ghost/instrumentation.rb, line 144
def start(name, parent: current, operation_id: SecureRandom.uuid, detached: false, diagnostic: nil, **attributes)
  previous = current unless detached
  validate_parent!(parent)
  parent_operation_id = parent.is_a?(Handle) ? parent.operation_id : parent
  payload = context.merge(attributes).merge(operation_id:, parent_operation_id:).compact
  payload = prepare_attributes(payload, diagnostic:)
  handle = Handle.new(
    self,
    name,
    operation_id:,
    parent_operation_id:,
    local_parent: parent.is_a?(Handle),
    previous:,
    payload:,
    started_at: monotonic_time,
    detached:
  )
  @mutex.synchronize do
    raise Error, "instrumentation is shut down" if @shutdown
    raise ArgumentError, "instrumentation operation is already active" if @handles.key?(operation_id)
    if parent.is_a?(Handle)
      raise Error, "instrumentation parent is not active" unless @handles[parent.operation_id].equal?(parent)
      raise Error, "instrumentation parent is finishing" if @finishing.key?(parent.operation_id)
    end

    @handles[operation_id] = handle
    @children[parent_operation_id] += 1 if parent.is_a?(Handle)
  end
  set_current(handle) unless detached
  notify(:start, handle.name, handle.payload)
  handle
rescue
  abandon(handle) if handle
  raise
end

# File lib/little_ghost/instrumentation.rb, line 106
def subscribe(subscriber, prepend: false)
  unless subscriber.is_a?(Subscriber)
    raise ArgumentError, "instrumentation subscriber must be a LittleGhost::Instrumentation::Subscriber"
  end

  @mutex.synchronize do
    @subscribers.reject! { |listener| listener.equal?(subscriber) }
    prepend ? @subscribers.unshift(subscriber) : @subscribers << subscriber
  end
  subscriber
end

# File lib/little_ghost/instrumentation.rb, line 260
def trace_context(**attributes)
  subscribers.each do |subscriber|
    value = subscriber.trace_context(**attributes)
    return value unless value.nil? || value.empty?
  rescue => error
    warn_failure(error, component: :subscriber)
  end
  {}
end

# File lib/little_ghost/instrumentation.rb, line 119
def unsubscribe(subscriber)
  @mutex.synchronize { @subscribers.reject! { |listener| listener.equal?(subscriber) } }
  subscriber
end

# File lib/little_ghost/instrumentation.rb, line 209
def with_context(attributes)
  values = deep_copy(context).merge(deep_copy(attributes.compact))
  ExecutionState.with(context_key => values) { yield }
end