Class: Insika::OutboxStore

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

Overview

Durable record of one pending OUTBOUND reply. A Shape B channel answers out of band — the turn ends and the reply is POSTed to the platform (or, for a relay, to the consumer's own callback) afterwards — so "the answer exists but the recipient is not on this connection" is exactly the problem already solved for async delegation. This store is DelegationStore's shape with a different recipient, deliberately: a second invention here would be a second thing to get wrong.

Lifecycle (never backwards):

pending     the turn committed a reply; nobody has claimed it yet. This is
          the ONLY status the boot sweep re-drives.
delivering  claimed. The claim happens BEFORE the HTTP call, so a crash
          here loses the delivery rather than duplicating it —
          at-most-once, the same honest scope as the delegation path.
delivered   the recipient answered 2xx.
failed      the bounded retry ran out. Terminal, and NOT re-driven at boot:
          a third party that refused N times is an operator problem, not
          something to replay forever.

failed and delivering are a deliberate widening of the two-status sketch (pending -> delivered): without them a crashed claim and an exhausted retry are indistinguishable from a fresh record, and the sweep would redeliver both.

Normalizes symbol->string on WRITE (the backend only round-trips JSON types), like every other domain store.

Defined Under Namespace

Classes: Delivery

Constant Summary collapse

SCOPE = 
"outbox"
KEY_PREFIX = 
"outbox:"
STATUSES = 
%i[pending delivering delivered failed].freeze

Instance Method Summary collapse

Constructor Details

#initialize(store:) ⇒ OutboxStore

Returns a new instance of OutboxStore.



44
45
46
 
# File 'lib/insika/outbox_store.rb', line 44

def initialize(store:)
  @store = store
end

Instance Method Details

#claim(id) ⇒ Object

pending -> delivering, ATOMICALLY — across processes, not just fibers: the read-check-write rides Store#transaction, so two workers claiming the same record serialize on the backend's lock and only one sees :pending. Returns true only for the caller that won the transition; that caller (and only it) makes the HTTP call, so delivery is at-most-once even if the terminal hook and the boot sweep both fire.



95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/insika/outbox_store.rb', line 95

def claim(id)
  @store.transaction do
    record = @store.get(SCOPE, key_for(id))
    next false unless record && record["status"] == "pending"

    record["status"] = "delivering"
    record["updated_at"] = timestamp
    @store.set(SCOPE, key_for(id), record)
    true
  end
end

#create(channel:, to:, task_id:, session_id:, payload:, index: 0, id: SecureRandom.uuid) ⇒ Object

-> Delivery (:pending). payload is the body the channel will send; it is DATA (string keys, JSON types) and the store never interprets it. index is the balloon's position inside its turn — 0 for a plain :at_end delivery, written by a progressive flush.



52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/insika/outbox_store.rb', line 52

def create(channel:, to:, task_id:, session_id:, payload:, index: 0, id: SecureRandom.uuid)
  record = {
    "id" => id.to_s,
    "channel" => channel.to_s,
    "to" => to.to_s,
    "task_id" => task_id&.to_s,
    "session_id" => session_id&.to_s,
    "payload" => payload,
    "status" => "pending",
    "attempts" => 0,
    "last_error" => nil,
    "index" => index.to_i,
    "created_at" => timestamp,
    "updated_at" => timestamp
  }
  @store.set(SCOPE, key_for(id), record)
  to_delivery(record)
end

#delete_older_than(time) ⇒ Object

WS8 retention: deliveries created before the cutoff. TERMINAL ones only — a pending/delivering record older than the window is still somebody's undelivered answer, and the sweep is not the place to decide it is lost. -> count removed.



152
153
154
155
156
157
 
# File 'lib/insika/outbox_store.rb', line 152

def delete_older_than(time)
  cutoff = time.utc.iso8601
  delete_where do |d|
    %i[delivered failed].include?(d.status) && d.created_at.to_s < cutoff
  end
end

#find(id) ⇒ Object

-> Delivery | nil



72
73
74
75
 
# File 'lib/insika/outbox_store.rb', line 72

def find(id)
  record = @store.get(SCOPE, key_for(id))
  record && to_delivery(record)
end

#mark_delivered(id) ⇒ Object

-> Delivery (:delivered). Idempotent: an already-delivered record is returned unchanged rather than re-marked.



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

def mark_delivered(id)
  record = fetch!(id)
  return to_delivery(record) if record["status"] == "delivered"

  record["status"] = "delivered"
  record["attempts"] = record["attempts"].to_i + 1
  record["last_error"] = nil
  touch(id, record)
end

#mark_failed(id, error: nil) ⇒ Object

-> Delivery (:failed). The retry budget is spent; nothing re-drives this.



130
131
132
133
134
135
 
# File 'lib/insika/outbox_store.rb', line 130

def mark_failed(id, error: nil)
  record = fetch!(id)
  record["status"] = "failed"
  record["last_error"] = error&.to_s if error
  touch(id, record)
end

#pending ⇒ Object

-> [Delivery] still waiting for a first claim (boot sweep). Deliberately NOT delivering: that one was claimed by a process that then died, and whether its POST landed is unknowable — replaying it is the duplicate the claim exists to prevent.

Ordered by [task_id, index] : a crashed progressive turn re-drives balloon 1 only after balloon 0, never the reverse.



84
85
86
87
 
# File 'lib/insika/outbox_store.rb', line 84

def pending
  scan.select { |d| d.status == :pending }
      .sort_by { |d| [d.task_id.to_s, d.index] }
end

#purge_sessions(session_ids) ⇒ Object

WS8 (LGPD): drops every delivery of these sessions, whatever its status. payload is the ANSWER as it was handed to the channel, so a purge that stops at the session record leaves the conversation readable here forever. -> count removed.



141
142
143
144
145
146
 
# File 'lib/insika/outbox_store.rb', line 141

def purge_sessions(session_ids)
  wanted = Array(session_ids).map(&:to_s)
  return 0 if wanted.empty?

  delete_where { |d| wanted.include?(d.session_id.to_s) }
end

#record_attempt(id, error: nil) ⇒ Object

One attempt happened and did not succeed. Keeps the record :delivering (the in-process retry loop owns it) and writes down why, so an operator reading the store sees the third party's answer and not just a counter.



110
111
112
113
114
115
 
# File 'lib/insika/outbox_store.rb', line 110

def record_attempt(id, error: nil)
  record = fetch!(id)
  record["attempts"] = record["attempts"].to_i + 1
  record["last_error"] = error&.to_s
  touch(id, record)
end