Class: Insika::ShadowPairStore

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

Overview

C2 — one durable record per mirrored exchange (shadow mode), written by TWO INDEPENDENT HALVES: ours at the turn's terminal, the incumbent's at the mirror. Both land on the same key — a digest of (channel, external_id, event_id), deterministic and order-free — so the two writers converge without an index and without ordering assumptions.

It stores and it counts; it does not judge, does not fold a verdict, does not know what the criterion says.

Status transitions (never backwards):

            ┌─ record_incumbent ─┐
(nothing) ────┤                    ├─▶ open ─▶ complete ─▶ judged
            └─ record_ours ─────┘       └─▶ silent   (never judged)
                                        └─▶ incomplete (expire)

Defined Under Namespace

Classes: Pair

Constant Summary collapse

SCOPE = 
"shadow_pairs"
KEY_PREFIX = 
"pair:"
STATUSES = 
%i[open complete silent judged incomplete].freeze

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(store:) ⇒ ShadowPairStore

Returns a new instance of ShadowPairStore.



39
40
41
 
# File 'lib/insika/shadow_pair_store.rb', line 39

def initialize(store:)
  @store = store
end

Class Method Details

.key_for(channel:, external_id:, event_id:) ⇒ Object

The correlation key BOTH writers compute independently. SHA-256 hex of "\0<external_id>\0<event_id>", truncated to 32 — deterministic, order-free, and it keeps a phone number out of the store's key space.



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

def self.key_for(channel:, external_id:, event_id:)
  Digest::SHA256.hexdigest("#{channel}\0#{external_id}\0#{event_id}")[0, 32]
end

Instance Method Details

#counts(since: nil) ⇒ Object

-> { open:, complete:, silent:, judged:, incomplete: }



118
119
120
121
 
# File 'lib/insika/shadow_pair_store.rb', line 118

def counts(since: nil)
  pairs = since ? self.since(since) : each.to_a
  STATUSES.to_h { |s| [s, pairs.count { |p| p.status == s }] }
end

#delete_older_than(time) ⇒ Object

Retention: pairs created before the cutoff, TERMINAL statuses only (judged/incomplete) — an open/complete record older than the window is still someone's unjudged evidence. -> count removed.



179
180
181
182
183
184
185
186
 
# File 'lib/insika/shadow_pair_store.rb', line 179

def delete_older_than(time)
  cutoff = time.utc.iso8601
  doomed = each.select do |p|
    %i[judged incomplete].include?(p.status) && p.created_at.to_s < cutoff
  end
  doomed.each { |p| @store.delete(SCOPE, key_for(p.id)) }
  doomed.size
end

#each(&block) ⇒ Object

Lazy scan over a SNAPSHOT of the keys (deleting under a live enumeration would skip records — the same rule OutboxStore applies).



91
92
93
94
95
96
97
98
 
# File 'lib/insika/shadow_pair_store.rb', line 91

def each(&block)
  return enum_for(:each) unless block_given?

  @store.list(SCOPE, KEY_PREFIX).each do |key|
    record = @store.get(SCOPE, key)
    yield to_pair(record) if record
  end
end

#expire(older_than:) ⇒ Object

An open pair older than the cutoff will never complete. -> count moved. complete/silent/judged are never touched. Update-style only: a pair deleted between the scan and the write (retention, LGPD purge) is left deleted — an upsert here would resurrect it as a ghost :incomplete record carrying none of its fields.



143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/insika/shadow_pair_store.rb', line 143

def expire(older_than:)
  cutoff = older_than.utc.iso8601
  moved = 0
  each.select { |p| p.status == :open && p.created_at.to_s < cutoff }.each do |pair|
    @store.transaction do
      key = key_for(pair.id)
      record = @store.get(SCOPE, key)
      next unless record && record["status"] == "open"

      record["status"] = "incomplete"
      record["updated_at"] = timestamp
      @store.set(SCOPE, key, record)
      moved += 1
    end
  end
  moved
end

#find(id) ⇒ Object

-> Pair | nil



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

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

#purge_sessions(session_ids) ⇒ Object

LGPD / retention (C9): drops every pair of these sessions, whatever its status — the pair holds the customer's own words. -> count removed.



167
168
169
170
171
172
173
174
 
# File 'lib/insika/shadow_pair_store.rb', line 167

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

  doomed = each.select { |p| wanted.include?(p.session_id.to_s) }
  doomed.each { |p| @store.delete(SCOPE, key_for(p.id)) }
  doomed.size
end

#record_incumbent(id:, channel:, event_id:, external_id:, reply:, at: nil) ⇒ Object

The incumbent's half (the mirror contract). Same upsert shape; the fields this half owns are the reply and, on first write, the timestamp the mirror reports. Never overwrites our half's fields. First-write-wins is enforced HERE, inside the transaction: the customer received ONE reply, and two concurrent mirror retries must not let the second rewrite the evidence. -> Pair



74
75
76
77
78
79
80
81
 
# File 'lib/insika/shadow_pair_store.rb', line 74

def record_incumbent(id:, channel:, event_id:, external_id:, reply:, at: nil)
  upsert(id, at: at) do |record, created|
    record["channel"] = channel.to_s
    record["event_id"] = event_id.to_s
    record["incumbent_reply"] = reply.to_s if record["incumbent_reply"].nil?
    created
  end
end

#record_ours(id:, channel:, agent:, session_id:, task_id:, event_id:, inbound:, reply:, criterion_sha:) ⇒ Object

Our half. Upsert: creates the record or fills our fields on the incumbent's. reply may be "" — a turn that published nothing (halt_when, an out-of-band tool) is recorded as :silent rather than left invisible. -> Pair



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

def record_ours(id:, channel:, agent:, session_id:, task_id:, event_id:,
                inbound:, reply:, criterion_sha:)
  upsert(id) do |record, created|
    record["channel"] = channel.to_s
    record["event_id"] = event_id.to_s
    record["agent"] = agent
    record["session_id"] = session_id&.to_s
    record["task_id"] = task_id&.to_s
    record["inbound"] = inbound.to_s
    record["insika_reply"] = reply.to_s
    record["criterion_sha"] = criterion_sha
    created
  end
end

#record_verdict(id, verdict:) ⇒ Object

The panel's Verdict as data. status -> :judged. -> Pair



124
125
126
127
128
129
130
131
132
133
134
135
136
 
# File 'lib/insika/shadow_pair_store.rb', line 124

def record_verdict(id, verdict:)
  @store.transaction do
    key = key_for(id)
    record = @store.get(SCOPE, key)
    raise Insika::NotFoundError, "shadow pair not found: #{id}" unless record

    record["verdict"] = Coercion.deep_stringify(verdict)
    record["status"] = "judged"
    record["updated_at"] = timestamp
    @store.set(SCOPE, key, record)
    to_pair(record)
  end
end

#since(time) ⇒ Object

-> [Pair] created at or after time.



101
102
103
104
 
# File 'lib/insika/shadow_pair_store.rb', line 101

def since(time)
  cutoff = time.utc.iso8601
  each.select { |p| p.created_at.to_s >= cutoff }
end

#size ⇒ Object

-> Integer. Keys only, no record materialization — a count of pairs must not pay for customer text (doctor's shadow-off check, the Studio).



163
 
# File 'lib/insika/shadow_pair_store.rb', line 163

def size = @store.list(SCOPE, KEY_PREFIX).length

#unjudged(limit: nil, agent: nil) ⇒ Object

-> [Pair] status :complete, oldest first — the judging queue. silent pairs are NEVER here: finding that pairwise is systematically unfair to a tool that delivers out of band is not something to average away.



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

def unjudged(limit: nil, agent: nil)
  pairs = each.select { |p| p.status == :complete }
               .sort_by { |p| p.created_at.to_s }
  pairs = pairs.select { |p| p.agent.to_s == agent.to_s } if agent
  limit ? pairs.first(limit.to_i) : pairs
end