Class: Insika::AlertDispatcher

Inherits:
Object
  • Object
show all
Defined in:
lib/insika/alert_dispatcher.rb

Overview

Operator alerts to a webhook (WS6): the events :budget_warning, :breaker_open and :delivery_failed are answered per AGENT — a profile declaring alerts: { "webhook" => url } gets its alerts POSTed there as JSON. The delivery reuses the outbox + claim mechanism whole (ChannelDelivery): the handler only WRITES the outbox row; the existing tick sweep and boot recovery claim and POST it, at-most-once with bounded retry, via a registered Channels::Webhook. The engine transports the event and does not interpret it — a Slack/CRM adapter is the consumer's.

Started as a child of the turn supervisor (like the tick) in serving mode; tests drive handle directly.

Constant Summary collapse

ALERT_TYPES = 
%i[budget_warning breaker_open delivery_failed].freeze

Instance Method Summary collapse

Constructor Details

#initialize(event_stream:, outbox:, channels:, profiles:, task_store: nil, http:) ⇒ AlertDispatcher

Returns a new instance of AlertDispatcher.



21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/insika/alert_dispatcher.rb', line 21

def initialize(event_stream:, outbox:, channels:, profiles:, task_store: nil, http:)
  @event_stream = event_stream
  @outbox = outbox
  @channels = channels
  @profiles = profiles
  @task_store = task_store
  @http = http
  @webhook_ids = {} # url -> registered channel id (one webhook per URL)
  # WS6 (boot recovery): webhook channels are derived from PROFILE config,
  # not from events. Registering lazily (on the first alert) means a pending
  # outbox row a crashed process left is swept at boot against an EMPTY
  # registry and marked failed terminal. Pre-registering every configured
  # URL at wiring time lets the boot sweep find the channel and deliver.
  register_all_webhooks
end

Instance Method Details

#handle(event) ⇒ Object

The event -> outbox row. Cheap (one transactional write); the DELIVERY is the tick's job. Never raises: an alerting failure must not break the turn.



57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/insika/alert_dispatcher.rb', line 57

def handle(event)
  type = event.type.to_s.to_sym
  return unless ALERT_TYPES.include?(type)

  agent = agent_for(event)
  return if agent.nil?

  profile = @profiles.respond_to?(:fetch) ? @profiles.fetch(agent.to_s) : nil
  return if profile.nil?

  url = profile&.respond_to?(:alerts) ? profile.alerts&.dig("webhook") : nil
  return if Coercion.blank?(url)

  record_alert(agent: agent.to_s, url: url.to_s, event: event)
rescue StandardError
  nil
end

#start(parent:) ⇒ Object

Serving: a long-lived consumer that answers every alert event. Drains on the supervisor fiber (blocks on the queue — no spin), exactly like the tick. It subscribes TYPED (only the alert events enter its queue — it answers payloads a full-traffic stream would otherwise overflow away) and, on an overflow close, RE-SUBSCRIBES: a consumer that never re-binds is how alerts stop in silence (WS6).



43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/insika/alert_dispatcher.rb', line 43

def start(parent:)
  parent.async do |t|
    t.annotate("insika-alerts")
    loop do
      subscription = @event_stream.subscribe(types: ALERT_TYPES)
      subscription.each { |event| handle(event) }
      # the subscription closed (its overflow path) — alerts must not die here
    end
  end
  true
end