Class: Insika::ChannelDelivery

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

Overview

Hands a finished turn's answer to a Shape B channel. The turn ended; the recipient is not on any connection; the reply has to travel out of band and survive a crash on the way. Three moves, in this order, and the order is the whole design:

1. RECORD at the turn's terminal (durable, `pending`).
2. CLAIM before the HTTP call (`pending -> delivering`, atomic). A crash
 between the claim and the POST loses that delivery; it does not duplicate
 it. At-most-once, stated rather than papered over — the same honest scope
 the async-delegation path already has.
3. RETRY, bounded and explicit. Unlike a delegation, the recipient is a third
 party with outages, so "keep trying" is a real requirement and "keep trying
 forever" is a real outage of ours.

It is NOT a job queue: no scheduler, no priorities, no fan-out. The moment it grows one, the thing to do is take a real queue, not to finish building this.

Constant Summary collapse

MAX_ATTEMPTS = 
3
BACKOFF_SECONDS =

Waits BETWEEN attempts, so attempt 1 is immediate. Short on purpose: a customer waiting on WhatsApp is the deadline, not the consumer's SLA.

[1, 5].freeze

Instance Method Summary collapse

Constructor Details

#initialize(channels:, outbox:, session_store:, event_stream: nil, max_attempts: MAX_ATTEMPTS, backoff: BACKOFF_SECONDS, sleeper: nil) ⇒ ChannelDelivery

Returns a new instance of ChannelDelivery.



28
29
30
31
32
33
34
35
36
37
 
# File 'lib/insika/channel_delivery.rb', line 28

def initialize(channels:, outbox:, session_store:, event_stream: nil,
               max_attempts: MAX_ATTEMPTS, backoff: BACKOFF_SECONDS, sleeper: nil)
  @channels = channels
  @outbox = outbox
  @session_store = session_store
  @event_stream = event_stream
  @max_attempts = max_attempts
  @backoff = Array(backoff)
  @sleeper = sleeper || method(:default_sleep)
end

Instance Method Details

#deliver(id) ⇒ Object

Claim + POST + bounded retry. Safe to call twice: the second caller loses the claim and returns without touching the recipient.



64
65
66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/insika/channel_delivery.rb', line 64

def deliver(id)
  return false unless @outbox.claim(id)

  delivery = @outbox.find(id)
  channel = @channels&.find(delivery&.channel)
  # The channel vanished between the record and the dispatch (a plugin was
  # disabled, the deployment was reconfigured). Nothing can send this; leave it
  # terminal so the boot sweep does not spin on it forever.
  if channel.nil? || !channel.respond_to?(:deliver)
    return finish(@outbox.mark_failed(id, error: "channel '#{delivery&.channel}' is not registered"))
  end

  attempt(delivery, channel)
end

#record(task:, channel_id:, content:) ⇒ Object

The turn committed an answer. -> the Delivery to dispatch, or nil when there is nothing to deliver, which is the common case and must stay cheap:

· the turn did not come in through a channel,
· the channel is Shape A (answers on its own stream — no `deliver`),
· the answer is empty (a turn that died mid-message published nothing, and
half a sentence was never an answer),
· or we do not know who to send it to.


46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/insika/channel_delivery.rb', line 46

def record(task:, channel_id:, content:)
  return nil if content.to_s.strip.empty?

  channel = @channels&.find(channel_id)
  return nil unless channel.respond_to?(:deliver)

  to = recipient(channel, task.session_id)
  return nil if to.nil? || to.empty?

  @outbox.create(
    channel: channel_id, to: to, task_id: task.id, session_id: task.session_id,
    payload: { "session_id" => task.session_id.to_s, "task_id" => task.id.to_s,
               "content" => content.to_s }
  )
end

#sweep ⇒ Object

Boot: re-drive what a previous process recorded and never claimed. Records left delivering are deliberately NOT swept — that process may have POSTed before it died, and replaying is the duplicate the claim exists to prevent. -> { dispatched: [ids] }



83
84
85
86
87
88
89
 
# File 'lib/insika/channel_delivery.rb', line 83

def sweep
  dispatched = @outbox.pending.map do |delivery|
    deliver(delivery.id)
    delivery.id
  end
  { dispatched: dispatched }
end