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 Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

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

Returns a new instance of ChannelDelivery.



28
29
30
31
32
33
34
35
36
37
38
39
40
 
# 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,
               shadow_pairs: nil, criterion_sha: 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)
  @shadow_pairs = shadow_pairs
  @criterion_sha = criterion_sha
end

Instance Attribute Details

#criterion_sha=(value) ⇒ Object (writeonly)

the pair store and the frozen criterion's sha. Both default to nil (parity — a graph without them behaves exactly as today); the server root sets them at boot, after the criterion file has been loaded and refused-or-accepted (the graph itself reads no env and no file).



46
47
48
 
# File 'lib/insika/channel_delivery.rb', line 46

def criterion_sha=(value)
  @criterion_sha = value
end

#shadow_pairs=(value) ⇒ Object (writeonly)

the pair store and the frozen criterion's sha. Both default to nil (parity — a graph without them behaves exactly as today); the server root sets them at boot, after the criterion file has been loaded and refused-or-accepted (the graph itself reads no env and no file).



46
47
48
 
# File 'lib/insika/channel_delivery.rb', line 46

def shadow_pairs=(value)
  @shadow_pairs = value
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.



102
103
104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/insika/channel_delivery.rb', line 102

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

#progressive?(channel_id) ⇒ Boolean

does this channel flush progressively? Duck-typed — a channel that does not answer progressive? is :at_end.

Returns:

  • (Boolean) 


95
96
97
98
 
# File 'lib/insika/channel_delivery.rb', line 95

def progressive?(channel_id)
  channel = @channels&.find(channel_id)
  channel.respond_to?(:progressive?) && channel.progressive?
end

#record_balloons(task:, channel_id:, content:, progressive:, attachments: nil) ⇒ Object

Confirmed answer -> 0..N pending Deliveries, in order . A progressive channel splits on paragraphs (BalloonSplitter); everything else is the single whole-answer row. -> [] when there is nothing to send (the cheap exits):

· 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),
· the channel is in SHADOW mode: the answer is recorded as a
pair and nothing is dispatched — zero outbox writes, ever (E1),
· or we do not know who to send it to.

attachments (evidence cards) ride the outbox payload as an ADDITIVE key on the LAST balloon — a Shape B channel that reads payload ignores it (JSON contract, additive); one that renders cards consumes it.



63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/insika/channel_delivery.rb', line 63

def record_balloons(task:, channel_id:, content:, progressive:, attachments: nil)
  channel = @channels&.find(channel_id)
  return [] unless channel.respond_to?(:deliver)
  # Shadow records ONE pair for the whole answer — a balloon per paragraph
  # would mint N pairs for one turn.
  if shadow?(channel)
    record_shadow(task, channel_id, content)
    return []
  end

  return [] if content.to_s.strip.empty?

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

  parts = progressive ? Insika::BalloonSplitter.split(content) : [content.to_s]
  parts = parts.reject { |p| p.to_s.strip.empty? }
  return [] if parts.empty?

  multi = parts.size > 1
  parts.each_with_index.map do |part, i|
    last = i == parts.size - 1
    create_pending(task, channel_id, part, to,
                   index: multi ? i : nil, final: multi ? last : nil,
                   attachments: last ? attachments : nil)
  end
end

#shadow?(channel) ⇒ Boolean

Returns:

  • (Boolean) 


91
 
# File 'lib/insika/channel_delivery.rb', line 91

def shadow?(channel) = channel.respond_to?(:shadow?) && channel.shadow?

#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] }



121
122
123
124
125
126
127
 
# File 'lib/insika/channel_delivery.rb', line 121

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