1# frozen_string_literal: true
2
3require "date"
4require "value_semantics/monkey_patched"
5require "em_promise"
6
7require_relative "blather_notify"
8require_relative "utils"
9
10class PortingStepRepo
11 # Any thing that debounces messages must happen inside this.
12 # The porting logic will pass everything in every time.
13 class Outputs
14 def info(port_id, kind, msg); end
15
16 def warn(port_id, kind, msg); end
17
18 def error(port_id, kind, e_or_msg); end
19
20 def to_customer(port_id, kind, tel, msg); end
21 end
22
23 value_semantics do
24 redis Anything(), default: LazyObject.new { REDIS }
25 blather_notify Anything(), default: BlatherNotify
26 admin_server Anything(), default: CONFIG[:admin_server]
27 testing_tel Anything(), default: CONFIG[:testing_tel]
28 output Outputs, default: Outputs.new
29 end
30
31 def find(port)
32 if_processing(port) do
33 case port.processing_status
34 when "FOC"
35 FOC.new(**to_h).find(port)
36 when "COMPLETE"
37 Complete.new(**to_h).find(port)
38 else
39 Wait.new(port, output: output)
40 end
41 end
42 end
43
44 def if_processing(port)
45 redis.exists("jmp_port_freeze-#{port.id}").then do |v|
46 next Frozen.new(port, output: output) if v == 1
47
48 yield
49 end
50 end
51
52 class FOC < self
53 # This is how long we'll wait for a port to move from FOC to COMPLETED
54 # It's in fractional days because DateTime
55 GRACE_PERIOD = 15.0 / (24 * 60)
56
57 def find(port)
58 Alert.for(
59 port,
60 grace_period: GRACE_PERIOD,
61 output: output, key: :late_foc,
62 msg: "⚠ Port is still in FOC state a while past FOC",
63 real_step: Wait.new(port, output: output)
64 )
65 end
66 end
67
68 class Complete < self
69 # If it's been 35 minutes and the number isn't reachable, a human
70 # should get involved
71 GRACE_PERIOD = 35.0 / (24 * 60)
72
73 def find(port)
74 if_not_complete(port) do
75 exe = blather_notify.command_execution(admin_server, "customer info")
76 AdminCommand.new(exe: exe, **to_h).find(port).then do |step|
77 Alert.for(
78 port,
79 grace_period: GRACE_PERIOD, output: output,
80 key: :late_finish, msg: msg(port), real_step: step
81 )
82 end
83 end
84 end
85
86 def if_not_complete(port)
87 redis.exists("jmp_port_complete-#{port.id}").then do |v|
88 next Done.new(port, output: output) if v == 1
89
90 yield
91 end
92 end
93
94 def msg(port)
95 "⚠ Port still hasn't finished. We'll keep trying unless you set redis " \
96 "key `jmp_port_freeze-#{port.id}`"
97 end
98
99 class NoCustomer < self
100 attr_reader :port
101
102 NO_GRACE_PERIOD = 0
103
104 def initialize(port:, **kwargs)
105 @port = port
106 super(**kwargs)
107 end
108
109 def self.for(port:, **kwargs)
110 Alert.for(
111 port,
112 grace_period: NO_GRACE_PERIOD,
113 real_step: new(port: port, **kwargs),
114 output: kwargs[:output],
115 msg: msg(port),
116 key: :port_for_unknown_customer
117 )
118 end
119
120 def self.msg(port)
121 "⚠ Freezing port #{port.id} for unknown customer: #{port.customer_id}."
122 end
123
124 def perform_next_step
125 redis.set("jmp_port_freeze-#{port.id}", 1)
126 end
127 end
128
129 class AdminCommand < self
130 def initialize(exe:, **kwargs)
131 @exe = exe
132 super(**kwargs)
133 end
134
135 def to_h
136 super.merge(exe: @exe)
137 end
138
139 def find(port)
140 @exe.fetch_and_submit(q: port.customer_id).then do |form|
141 next NoCustomer.for(port: port, **to_h.except(:exe)) unless form.tel
142
143 if port.tel == form.tel
144 GoodNumber.new(**to_h).find(port)
145 else
146 WrongNumber.new(port: port, execution: @exe)
147 end
148 end
149 end
150
151 class GoodNumber < self
152 def find(port)
153 Reachability.new(type: "voice", **to_h).test(port) do
154 Reachability.new(type: "sms", **to_h).test(port) do
155 FinishUp.new(port, redis: redis, output: output)
156 end
157 end
158 rescue StandardError => e
159 ReachabilityFailure.for(error: e, port: port, **to_h)
160 end
161
162 class ReachabilityFailure < self
163 attr_reader :error
164
165 NO_GRACE_PERIOD = 0
166
167 # @param error [StandardError]
168 def self.for(port:, error:, **kwargs)
169 Alert.for(
170 port,
171 grace_period: NO_GRACE_PERIOD,
172 real_step: nil,
173 output: kwargs[:output],
174 msg: msg(port, error),
175 key: :reachability_failure
176 )
177 end
178
179 def self.msg(port, error)
180 "⚠ Error checking #{port.id} reachability: #{error}"
181 end
182 end
183
184 class Reachability < self
185 def initialize(type:, **args)
186 @type = type
187 super(**args)
188 end
189
190 def test(port)
191 execute_command(port).then { |response|
192 next yield unless response.count == "0"
193
194 args = to_h.slice(
195 :blather_notify, :testing_tel, :admin_server, :output
196 )
197 RunTest.new(type: @type, tel: port.tel, port: port, **args)
198 }.catch do |e|
199 ReachabilityFailure.for(port: port, error: e, output: output)
200 end
201 end
202
203 class RunTest
204 # This is here for tests
205 attr_reader :type
206
207 def initialize(
208 type:, tel:, port:, output:,
209 blather_notify:, testing_tel:, admin_server:
210 )
211 @type = type
212 @tel = tel
213 @port = port
214 @output = output
215 @blather_notify = blather_notify
216 @testing_tel = testing_tel
217 @admin_server = admin_server
218 end
219
220 def perform_next_step
221 @blather_notify.command_execution(@admin_server, "reachability")
222 .fetch_and_submit(
223 tel: @tel, type: @type, reachability_tel: @testing_tel
224 ).catch do |e|
225 ReachabilityFailure.for(
226 port: @port, error: e, output: @output
227 ).perform_next_step
228 end
229 end
230 end
231
232 protected
233
234 def execute_command(port)
235 blather_notify.command_execution(admin_server, "reachability")
236 .fetch_and_submit(tel: port.tel, type: @type)
237 end
238 end
239
240 class FinishUp
241 MESSAGE = "Hi! This is JMP support - your number has " \
242 "successfully transferred in to JMP! All calls/messages " \
243 "will now use your transferred-in number - your old JMP " \
244 "number has been disabled. Let us know if you have any " \
245 "questions and thanks for using JMP!"
246
247 def initialize(port, redis:, output:)
248 @port = port
249 @redis = redis
250 @output = output
251 end
252
253 def set_key
254 @redis.set(
255 "jmp_port_complete-#{@port.id}",
256 DateTime.now.iso8601,
257 "EX",
258 60 * 60 * 24 * 2 ### 2 Days should be enough to not see it listed
259 )
260 end
261
262 def perform_next_step
263 set_key.then do
264 EMPromise.all([
265 @output.info(@port.id, :complete, "Port Complete!"),
266 @output.to_customer(
267 @port.id, :complete,
268 Tel.new(@port.tel), MESSAGE
269 )
270 ])
271 end
272 end
273 end
274 end
275
276 class WrongNumber
277 attr_reader :new_backend
278
279 def initialize(port:, execution:)
280 @new_backend = port.backend_sgx
281 @right_number = port.tel
282 @exe = execution
283 end
284
285 def perform_next_step
286 @exe.fetch_and_submit(action: "number_change").then do |_form|
287 @exe.fetch_and_submit(
288 new_backend: @new_backend,
289 new_tel: @right_number,
290 should_delete: "true"
291 )
292 end
293 end
294 end
295 end
296 end
297
298 # This doesn't do anything and just waits for something to happen later
299 class Wait
300 def initialize(port, output:)
301 @port = port
302 @output = output
303 end
304
305 def perform_next_step
306 @output.info(@port.id, :wait, "Waiting...")
307 end
308 end
309
310 # This also doesn't do anything but is more content about it
311 class Done
312 def initialize(port, output:)
313 @port = port
314 @output = output
315 end
316
317 def perform_next_step
318 @output.info(@port.id, :done, "Done.")
319 end
320 end
321
322 # This also also doesn't do anything but is intentional
323 class Frozen
324 def initialize(port, output:)
325 @port = port
326 @output = output
327 end
328
329 def perform_next_step
330 @output.info(@port.id, :frozen, "Frozen.")
331 end
332 end
333
334 # This class sends and error to the human to check things out
335 class Alert
336 # @param [PortingStepRepo, NilClass] real_step If `nil`, just notify
337 def self.for(port, grace_period:, real_step:, **args)
338 if (DateTime.now - port.actual_foc_date.to_datetime) > grace_period
339 new(port, real_step: real_step, **args)
340 else
341 real_step
342 end
343 end
344
345 # For tests
346 attr_reader :key
347 attr_reader :real_step
348
349 def initialize(port, real_step:, output:, msg:, key:)
350 @port = port
351 @real_step = real_step
352 @output = output
353 @msg = msg
354 @key = key
355 end
356
357 def perform_next_step
358 @output.warn(@port.id, @key, @msg).then {
359 @real_step&.perform_next_step
360 }
361 end
362 end
363end
364
365class FullManual < PortingStepRepo::Outputs
366 def info(id, key, msg)
367 puts "[#{id}] INFO(#{key}): #{msg}"
368 end
369
370 def warn(id, key, msg)
371 puts "[#{id}] WARN(#{key}): #{msg}"
372 end
373
374 def error(id, key, e_or_msg)
375 puts "[#{id}] ERRR(#{key}): #{e_or_msg}"
376 return unless e_or_msg.respond_to?(:backtrace)
377
378 e_or_msg.backtrace.each do |b|
379 puts "[#{id}] ERRR(#{key}): #{b}"
380 end
381 end
382
383 def to_customer(id, key, tel, msg)
384 puts "[#{id}] CUST(#{key}, #{tel}): #{msg}"
385 end
386end
387
388class ObservedAuto < FullManual
389 def initialize(endpoint, source_number)
390 @endpoint = endpoint
391 @src = source_number
392 end
393
394 def to_customer(id, key, tel, msg)
395 ExpiringLock.new(lock_key(id, key)).with do
396 EM::HttpRequest
397 .new(@endpoint)
398 .apost(
399 head: { "Content-Type" => "application/json" },
400 body: format_msg(tel, msg)
401 )
402 end
403 end
404
405protected
406
407 def lock_key(id, key)
408 "jmp_port_customer_msg_#{key}-#{id}"
409 end
410
411 def format_msg(tel, msg)
412 [{
413 time: DateTime.now.iso8601,
414 type: "message-received",
415 to: tel,
416 description: "Incoming message received",
417 message: actual_message(tel, msg)
418 }].to_json
419 end
420
421 def actual_message(tel, msg)
422 {
423 id: SecureRandom.uuid,
424 owner: tel,
425 applicationId: SecureRandom.uuid,
426 time: DateTime.now.iso8601,
427 segmentCount: 1,
428 direction: "in",
429 to: [tel], from: @src,
430 text: msg
431 }
432 end
433end
434
435class FullAutomatic < ObservedAuto
436 using FormToH
437
438 def initialize(pubsub_addr, endpoint, source_number)
439 @pubsub = BlatherNotify.pubsub(pubsub_addr)
440
441 Sentry.init do |config|
442 config.background_worker_threads = 0
443 end
444
445 super(endpoint, source_number)
446 end
447
448 # No one's watch; swallow informational messages
449 def info(*); end
450
451 def warn(id, key, msg)
452 ExpiringLock.new(warn_lock_key(id, key), expiry: 60 * 15).with do
453 entrykey = "#{id}:#{key}"
454 @pubsub.publish("#{entrykey}": error_entry("Port Warning", msg, entrykey))
455 end
456 end
457
458 def error(id, key, e_or_msg)
459 Sentry.with_scope do |scope|
460 scope.set_context("port", { id: id, action: key })
461
462 if e_or_msg.is_a?(::Exception)
463 Sentry.capture_exception(e_or_msg)
464 else
465 Sentry.capture_message(e_or_msg.to_s)
466 end
467 end
468 end
469
470protected
471
472 def error_entry(title, text, id)
473 Nokogiri::XML::Builder.new { |xml|
474 xml.entry(xmlns: "http://www.w3.org/2005/Atom") do
475 xml.updated DateTime.now.iso8601
476 xml.id id
477 xml.title title
478 xml.content text.to_s, type: "text"
479 xml.author { xml.name "porting" }
480 xml.generator "porting", version: "1.0"
481 end
482 }.doc.root
483 end
484
485 def warn_lock_key(id, key)
486 "jmp_port_warn_msg_#{key}-#{id}"
487 end
488end