1# frozen_string_literal: true
2
3require "sentry-ruby"
4require "statsd-instrument"
5
6require_relative "customer_repo"
7require_relative "session_manager"
8
9class Command
10 def self.execution
11 Thread.current[:execution]
12 end
13
14 def self.execution=(exe)
15 Thread.current[:execution] = exe
16 end
17
18 def self.reply(stanza=nil, &blk)
19 execution.reply(stanza, &blk)
20 end
21
22 def self.finish(*args, **kwargs, &blk)
23 execution.finish(*args, **kwargs, &blk)
24 end
25
26 def self.customer
27 execution.customer
28 end
29
30 def self.log
31 execution.log
32 end
33
34 class Execution
35 class Timeout < SessionManager::Timeout; end
36
37 class FinalStanza
38 attr_reader :stanza
39
40 def initialize(stanza)
41 @stanza = stanza
42 end
43 end
44
45 attr_reader :customer_repo, :log, :iq
46
47 def initialize(customer_repo, blather, format_error, iq)
48 @customer_repo = customer_repo
49 @blather = blather
50 @format_error = format_error
51 @iq = iq
52 @log = Thread.current[:log] || LOG
53 end
54
55 def execute
56 StatsD.increment("command", tags: ["node:#{iq.node}"])
57 EMPromise.resolve(nil).then {
58 Command.execution = self
59 catch_after(EMPromise.resolve(yield self))
60 }
61 end
62
63 def reply(stanza=nil)
64 stanza ||= iq.reply.tap { |reply| reply.status = :executing }
65 yield stanza if block_given?
66 COMMAND_MANAGER.write(stanza).then { |new_iq|
67 @iq = new_iq
68 }.catch_only(Blather::Stanza::Iq) { |new_iq|
69 @iq = new_iq if new_iq.set?
70 EMPromise.reject(new_iq)
71 }.catch_only(SessionManager::Timeout) do
72 EMPromise.reject(Timeout.new)
73 end
74 end
75
76 def finish(text=nil, type: :info, status: :completed)
77 reply = @iq.reply
78 reply.status = status
79 EMPromise.resolve(block_given? ? yield(reply) : nil).then {
80 if text
81 reply.note_type = type
82 reply.note_text = text
83 end
84 }.then do
85 EMPromise.reject(FinalStanza.new(reply))
86 end
87 end
88
89 def customer
90 @customer ||= @customer_repo.find_by_jid(@iq.from.stripped).then { |c|
91 Sentry.set_user(
92 id: c.customer_id,
93 jid: @iq.from.stripped
94 )
95 c
96 }
97 end
98
99 protected
100
101 def catch_after(promise)
102 promise.catch_only(Blather::Stanza::Iq::Command) { |iq|
103 next EMPromise.reject(iq) unless iq.cancel?
104
105 finish(status: :canceled)
106 }.catch_only(Timeout) { nil }.catch_only(FinalStanza) { |e|
107 @blather << e.stanza
108 }.catch do |e|
109 e = RuntimeError.new(e) if e.is_a?(String)
110 send_final_error(e)
111 EMPromise.reject(e)
112 end
113 end
114
115 def send_final_error(e)
116 def e.replied?
117 true
118 end
119
120 finish(@format_error.call(e), type: :error).catch_only(FinalStanza) do |s|
121 @blather << s.stanza
122 end
123 end
124 end
125
126 attr_reader :node, :name
127
128 def initialize(
129 node,
130 name,
131 customer_repo: CustomerRepo.new,
132 list_for: ->(tel:, **) { !!tel },
133 format_error: ->(e) { e.respond_to?(:message) ? e.message : e.to_s },
134 &blk
135 )
136 @node = node
137 @name = name
138 @customer_repo = customer_repo
139 @list_for = list_for
140 @format_error = format_error
141 @blk = blk
142 end
143
144 def register(blather, guards: [:execute?, { node: @node, sessionid: nil }])
145 blather.command(*guards) do |iq|
146 Execution.new(@customer_repo, blather, @format_error, iq).execute(&@blk)
147 end
148 self
149 end
150
151 def list_for?(**kwargs)
152 @list_for.call(**kwargs)
153 end
154end