Class: Insika::ScheduleStore

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

Overview

the per-agent schedule rows of the recurring feature — one row per declared schedule, keyed tenant:agent:id, holding BOTH the declaration (a copy of the profile's schedules entry — the engine reads the store, never the profile, on the fire path) and the runtime state (next_fire_at, last_run_at, last_task_id, last_skip).

The row is a DERIVED record: the profile declaration is the config (the DSL, the API, the Studio all edit the profile); the ScheduleEngine reconciles rows with declarations at each pass. The engine OWNS the runtime writes — transition_fire/mark_skip are called only inside the engine's transaction (the D5 discipline), never by a consumer. A row that is no longer declared is deleted (no zombie schedules); a row whose declaration is malformed is deleted too — a broken schedule must not keep firing with stale text (the doctor names it).

Multi-worker at-most-once rides Store#transaction: the pass re-reads the row inside the transaction, so two workers serialize on the backend lock and only one advances next_fire_at.

Defined Under Namespace

Classes: Record

Constant Summary collapse

SCOPE = 
"schedules"

Instance Method Summary collapse

Constructor Details

#initialize(store:) ⇒ ScheduleStore

Returns a new instance of ScheduleStore.



33
34
35
 
# File 'lib/insika/schedule_store.rb', line 33

def initialize(store:)
  @store = store
end

Instance Method Details

#all ⇒ Object

-> [Record] — every row (the Studio grid, the engine pass).



79
80
81
 
# File 'lib/insika/schedule_store.rb', line 79

def all
  @store.list(SCOPE).filter_map { |k| to_record(@store.get(SCOPE, k)) }
end

#delete(tenant:, agent:, id:) ⇒ Object 



112
113
114
 
# File 'lib/insika/schedule_store.rb', line 112

def delete(tenant:, agent:, id:)
  @store.delete(SCOPE, key(tenant, agent, id))
end

#due(now: Time.now.utc) ⇒ Object

-> [Record] — enabled AND due, oldest first (determinism). A row whose next_fire_at is nil (a cron that can never fire) is never due.



85
86
87
88
89
 
# File 'lib/insika/schedule_store.rb', line 85

def due(now: Time.now.utc)
  cutoff = now.iso8601
  all.select { |r| r.enabled && !r.next_fire_at.to_s.empty? && r.next_fire_at.to_s <= cutoff }
     .sort_by { |r| [r.next_fire_at.to_s, r.id] }
end

#find(tenant:, agent:, id:) ⇒ Object

-> Record | nil



63
64
65
66
 
# File 'lib/insika/schedule_store.rb', line 63

def find(tenant:, agent:, id:)
  record = @store.get(SCOPE, key(tenant, agent, id))
  record && to_record(record)
end

#for_agent(tenant:, agent:) ⇒ Object

-> [Record] — one agent's rows (the Studio's read), lexicographic.



69
70
71
72
73
74
75
76
 
# File 'lib/insika/schedule_store.rb', line 69

def for_agent(tenant:, agent:)
  prefix = "#{tenant_id(tenant)}:#{agent}:"
  @store.list(SCOPE).filter_map do |k|
    next unless k.start_with?(prefix)

    to_record(@store.get(SCOPE, k))
  end
end

#mark_skip(id:, tenant:, agent:, reason:, next_fire_at:, now: Time.now.utc) ⇒ Object

The visible skip — recorded, never silent: next_fire_at advances (the window is skipped, not queued) and last_skip names why.



105
106
107
108
109
110
 
# File 'lib/insika/schedule_store.rb', line 105

def mark_skip(id:, tenant:, agent:, reason:, next_fire_at:, now: Time.now.utc)
  mutate(tenant, agent, id, now) do |record|
    record["last_skip"] = { "at" => now.utc.iso8601, "reason" => reason.to_s }
    record["next_fire_at"] = next_fire_at.iso8601
  end
end

#purge(tenant:) ⇒ Object

-> count removed. The LGPD sweep (a tenant's schedules die with it).



117
118
119
120
121
122
 
# File 'lib/insika/schedule_store.rb', line 117

def purge(tenant:)
  prefix = "#{tenant_id(tenant)}:"
  keys = @store.list(SCOPE).select { |k| k.start_with?(prefix) }
  keys.each { |k| @store.delete(SCOPE, k) }
  keys.size
end

#sync_declared(tenant:, agent:, schedules: nil, now: Time.now.utc) ⇒ Object

Reconcile: upsert the declaration's row, or delete it when the declaration is gone/malformed. Called by the engine inside its pass transaction. A NEW or CHANGED declaration recomputes next_fire_at (the trigger lattice starts fresh); an unchanged row keeps its.



41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/insika/schedule_store.rb', line 41

def sync_declared(tenant:, agent:, schedules: nil, now: Time.now.utc)
  kept = []
  Array(schedules).each do |declaration|
    schedule = Insika::Schedule.parse(declaration)
    next if schedule.nil? # malformed — dropped, the doctor names it

    keep = upsert(tenant: tenant, agent: agent, schedule: schedule, now: now)
    kept << schedule.id if keep
  end

  # drop rows the agent no longer declares
  prefix = "#{tenant_id(tenant)}:#{agent}:"
  @store.list(SCOPE).each do |k|
    next unless k.start_with?(prefix)
    next if kept.include?(k.split(":").last)

    @store.delete(SCOPE, k)
  end
  kept
end

#transition_fire(id:, tenant:, agent:, task_id:, next_fire_at:, now: Time.now.utc) ⇒ Object

The engine's atomic fire claim — the LAST step of the fire's transaction: advance the lattice, stamp the run, clear the skip. Callers pass id and the engine's OWN transaction encloses this.



94
95
96
97
98
99
100
101
 
# File 'lib/insika/schedule_store.rb', line 94

def transition_fire(id:, tenant:, agent:, task_id:, next_fire_at:, now: Time.now.utc)
  mutate(tenant, agent, id, now) do |record|
    record["last_run_at"] = now.utc.iso8601
    record["last_task_id"] = task_id.to_s
    record["next_fire_at"] = next_fire_at.iso8601
    record["last_skip"] = nil
  end
end