6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
# File 'lib/textus/protocol/handlers/ops/pulse_coordinator.rb', line 6
def call(role:, ctx:, since: nil)
cursor_val = since || ctx.cursor_store.get(role)[:seq]
audit_filter = AuditFilter.new
audit_result = audit_filter.call(ctx:, filters: { seq_since: cursor_val })
changed = audit_result["rows"] || []
queue = ctx.manifest.policy.queue_lane
pending = queue ? ctx.list(lane: queue) : []
runner = Textus::Lanes::Knowledge::Runner.new(ctx.manifest, cursor_store: ctx.cursor_store)
result = {
"cursor" => ctx.audit_log.latest[:seq],
"changed" => changed,
"pending_review" => pending.is_a?(Array) ? pending.map { |r| r["key"] } : [],
"contract_etag" => Textus::Value::Etag.for_contract(ctx.layout),
"index_etag" => pulse_index_etag(ctx),
"loop_state" => runner.current_position(role),
}
ctx.cursor_store.set(role, result["cursor"])
result
end
|