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 yield reply if block_given?
80 if text
81 reply.note_type = type
82 reply.note_text = text
83 end
84 EMPromise.reject(FinalStanza.new(reply))
85 end
86
87 def customer
88 @customer ||= @customer_repo.find_by_jid(@iq.from.stripped).then { |c|
89 Sentry.set_user(
90 id: c.customer_id,
91 jid: @iq.from.stripped
92 )
93 c
94 }
95 end
96
97 protected
98
99 def catch_after(promise)
100 promise.catch_only(Blather::Stanza::Iq::Command) { |iq|
101 next EMPromise.reject(iq) unless iq.cancel?
102
103 finish(status: :canceled)
104 }.catch_only(Timeout) { nil }.catch_only(FinalStanza) { |e|
105 @blather << e.stanza
106 }.catch do |e|
107 send_final_error(e)
108 EMPromise.reject(e)
109 end
110 end
111
112 def send_final_error(e)
113 def e.replied?
114 true
115 end
116
117 finish(@format_error.call(e), type: :error).catch_only(FinalStanza) do |s|
118 @blather << s.stanza
119 end
120 end
121 end
122
123 attr_reader :node, :name
124
125 def initialize(
126 node,
127 name,
128 customer_repo: CustomerRepo.new,
129 list_for: ->(tel:, **) { !!tel },
130 format_error: ->(e) { e.respond_to?(:message) ? e.message : e.to_s },
131 &blk
132 )
133 @node = node
134 @name = name
135 @customer_repo = customer_repo
136 @list_for = list_for
137 @format_error = format_error
138 @blk = blk
139 end
140
141 def register(blather, guards: [:execute?, { node: @node, sessionid: nil }])
142 blather.command(*guards) do |iq|
143 Execution.new(@customer_repo, blather, @format_error, iq).execute(&@blk)
144 end
145 self
146 end
147
148 def list_for?(**kwargs)
149 @list_for.call(**kwargs)
150 end
151end