Module: Textus::Protocol::Handlers::System

Extended by:
System
Includes:
Concern
Included in:
System
Defined in:
lib/textus/protocol/handlers/system.rb

Instance Method Summary collapse

Methods included from Concern

included, #response

Instance Method Details

#audit(ctx:, call:, key: nil, lane: nil, role: nil, verb: nil, since: nil, seq_since: nil, correlation_id: nil, limit: nil) ⇒ Object 



23
24
25
26
 
# File 'lib/textus/protocol/handlers/system.rb', line 23

def audit(ctx:, call:, key: nil, lane: nil, role: nil, verb: nil, since: nil, seq_since: nil, correlation_id: nil, limit: nil)
  filters = { key:, lane:, role:, verb:, since:, seq_since:, correlation_id:, limit: }
  Ops::AuditFilter.new.call(ctx:, filters:)
end

#blame(key:, ctx:, call:) ⇒ Object 



28
29
30
31
32
33
 
# File 'lib/textus/protocol/handlers/system.rb', line 28

def blame(key:, ctx:, call:)
  audit_result = audit(key:, ctx:, call:)
  audit_rows = audit_result["rows"] || []
  git_map = Ops::GitBlame.new.call(key:, ctx:)
  audit_rows.map { |row| row.merge("git" => git_map ? git_map[row["ts"]] : nil) }
end

#boot(ctx:, call:) ⇒ Object 



8
9
10
11
12
13
14
15
16
17
 
# File 'lib/textus/protocol/handlers/system.rb', line 8

def boot(ctx:, call:)
  proxy = Textus::Protocol::Container.new(
    store: ctx.container.store,
    layout: ctx.layout,
    manifest: ctx.manifest, schemas: ctx.schemas,
    workflows: nil,
    root: ctx.layout.root
  )
  Textus::Protocol::Boot.build(container: proxy)
end

#drain(ctx:, call:) ⇒ Object 



60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/textus/protocol/handlers/system.rb', line 60

def drain(ctx:, call:)
  registry = ctx.workflow_registry || Textus::Workflow::Registry.new(ctx.workflows)
  reader = ->(key:) { ctx.read(key:) }

  Textus::Workflow::Scheduler.new(
    registry:, workflow_queue: ctx.workflow_queue, reader:,
  ).seed_expired

  consumer = Textus::Workflow::Consumer.new(registry:, container: ctx.container, call:)
  completed = 0
  failed = 0
  loop do
    row = ctx.workflow_queue.pop
    break unless row

    result = consumer.consume(row, ctx.workflow_queue)
    result[:state] == "done" ? completed += 1 : failed += 1
  end

  response(ok: failed.zero?, completed:, failed:)
end

#pulse(ctx:, call:, since: nil) ⇒ Object 



35
36
37
 
# File 'lib/textus/protocol/handlers/system.rb', line 35

def pulse(ctx:, call:, since: nil)
  Ops::PulseCoordinator.new.call(role: call.role, ctx:, since:)
end

#schema_list(ctx:, call:) ⇒ Object 



19
20
21
 
# File 'lib/textus/protocol/handlers/system.rb', line 19

def schema_list(ctx:, call:)
  Textus::Protocol::Boot.build_schemas(ctx.schemas)
end

#workflow_spec(name:, ctx:, call:) ⇒ Object 



39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/textus/protocol/handlers/system.rb', line 39

def workflow_spec(name:, ctx:, call:)
  registry = ctx.workflow_registry || Textus::Workflow::Registry.new(ctx.workflows)
  wf = registry.for(name)
  {
    "name" => wf.name, "type" => wf.workflow_type,
    "match_patterns" => wf.on_patterns || [],
    "handles_events" => wf.handled_events,
    "steps" => wf.steps.map { |s| { "name" => s.name.to_s, "type" => step_type(s) } },
    "publish_targets" => wf.publish_targets.map do |t|
      h = {}
      h["to"] = t.to if t.to
      h["template"] = t.template if t.template
      h
    end,
    "priority" => wf.priority,
    "max_attempts" => wf.max_attempts,
    "ttl" => wf.ttl&.to_s,
    "on_expire" => wf.on_expire
  }
end