1# frozen_string_literal: true
2
3require "pg/em/connection_pool"
4require "bigdecimal"
5require "blather/client/dsl" # Require this first to not auto-include
6require "blather/client"
7require "braintree"
8require "date"
9require "dhall"
10require "em-hiredis"
11require "em_promise"
12require "ruby-bandwidth-iris"
13require "sentry-ruby"
14require "statsd-instrument"
15
16Sentry.init
17
18CONFIG =
19 Dhall::Coder
20 .new(safe: Dhall::Coder::JSON_LIKE + [Symbol, Proc])
21 .load(ARGV[0], transform_keys: ->(k) { k&.to_sym })
22
23singleton_class.class_eval do
24 include Blather::DSL
25 Blather::DSL.append_features(self)
26end
27
28require_relative "lib/alt_top_up_form"
29require_relative "lib/add_bitcoin_address"
30require_relative "lib/backend_sgx"
31require_relative "lib/bandwidth_tn_order"
32require_relative "lib/btc_sell_prices"
33require_relative "lib/buy_account_credit_form"
34require_relative "lib/command_list"
35require_relative "lib/customer"
36require_relative "lib/electrum"
37require_relative "lib/em"
38require_relative "lib/payment_methods"
39require_relative "lib/registration"
40require_relative "lib/transaction"
41require_relative "lib/web_register_manager"
42
43ELECTRUM = Electrum.new(**CONFIG[:electrum])
44
45Faraday.default_adapter = :em_synchrony
46BandwidthIris::Client.global_options = {
47 account_id: CONFIG[:creds][:account],
48 username: CONFIG[:creds][:username],
49 password: CONFIG[:creds][:password]
50}
51
52def new_sentry_hub(stanza, name: nil)
53 hub = Sentry.get_current_hub&.new_from_top
54 raise "Sentry.init has not been called" unless hub
55
56 hub.push_scope
57 hub.current_scope.clear_breadcrumbs
58 hub.current_scope.set_transaction_name(name) if name
59 hub.current_scope.set_user(jid: stanza.from.stripped.to_s)
60 hub
61end
62
63# Braintree is not async, so wrap in EM.defer for now
64class AsyncBraintree
65 def initialize(environment:, merchant_id:, public_key:, private_key:, **)
66 @gateway = Braintree::Gateway.new(
67 environment: environment,
68 merchant_id: merchant_id,
69 public_key: public_key,
70 private_key: private_key
71 )
72 end
73
74 def respond_to_missing?(m, *)
75 @gateway.respond_to?(m)
76 end
77
78 def method_missing(m, *args)
79 return super unless respond_to_missing?(m, *args)
80
81 EM.promise_defer(klass: PromiseChain) do
82 @gateway.public_send(m, *args)
83 end
84 end
85
86 class PromiseChain < EMPromise
87 def respond_to_missing?(*)
88 false # We don't actually know what we respond to...
89 end
90
91 def method_missing(m, *args)
92 return super if respond_to_missing?(m, *args)
93 self.then { |o| o.public_send(m, *args) }
94 end
95 end
96end
97
98BRAINTREE = AsyncBraintree.new(**CONFIG[:braintree])
99
100def panic(e, hub=nil)
101 m = e.respond_to?(:message) ? e.message : e
102 warn "Error raised during event loop: #{e.class}: #{m}"
103 warn e.backtrace if e.respond_to?(:backtrace)
104 if e.is_a?(::Exception)
105 (hub || Sentry).capture_exception(e, hint: { background: false })
106 else
107 (hub || Sentry).capture_message(e.to_s, hint: { background: false })
108 end
109 exit 1
110end
111
112EM.error_handler(&method(:panic))
113
114when_ready do
115 BLATHER = self
116 REDIS = EM::Hiredis.connect
117 BTC_SELL_PRICES = BTCSellPrices.new(REDIS, CONFIG[:oxr_app_id])
118 DB = PG::EM::ConnectionPool.new(dbname: "jmp")
119 DB.type_map_for_results = PG::BasicTypeMapForResults.new(DB)
120 DB.type_map_for_queries = PG::BasicTypeMapForQueries.new(DB)
121
122 EM.add_periodic_timer(3600) do
123 ping = Blather::Stanza::Iq::Ping.new(:get, CONFIG[:server][:host])
124 ping.from = CONFIG[:component][:jid]
125 self << ping
126 end
127end
128
129# workqueue_count MUST be 0 or else Blather uses threads!
130setup(
131 CONFIG[:component][:jid],
132 CONFIG[:component][:secret],
133 CONFIG[:server][:host],
134 CONFIG[:server][:port],
135 nil,
136 nil,
137 workqueue_count: 0
138)
139
140message to: /\Aaccount@/, body: /./ do |m|
141 StatsD.increment("deprecated_account_bot")
142
143 self << m.reply.tap do |out|
144 out.body = "This bot is deprecated. Please talk to xmpp:cheogram.com"
145 end
146end
147
148before(
149 :iq,
150 type: [:error, :result],
151 to: /\Acustomer_/,
152 from: /(\A|@)#{CONFIG[:sgx]}(\/|\Z)/
153) { |iq| halt if IQ_MANAGER.fulfill(iq) }
154
155before nil, to: /\Acustomer_/, from: /(\A|@)#{CONFIG[:sgx]}(\/|\Z)/ do |s|
156 StatsD.increment("stanza_customer")
157
158 sentry_hub = new_sentry_hub(s, name: "stanza_customer")
159 Customer.for_customer_id(
160 s.to.node.delete_prefix("customer_")
161 ).then { |customer|
162 sentry_hub.current_scope.set_user(
163 id: customer.customer_id,
164 jid: s.from.stripped.to_s
165 )
166 customer.stanza_to(s)
167 }.catch { |e| panic(e, sentry_hub) }
168 halt
169end
170
171# Ignore messages to component
172# Especially if we have the component join MUC for notifications
173message(to: /\A#{CONFIG[:component][:jid]}\Z/) { true }
174
175def billable_message(m)
176 (m.body && !m.body.empty?) || m.find("ns:x", ns: OOB.registered_ns).first
177end
178
179message do |m|
180 StatsD.increment("message")
181
182 sentry_hub = new_sentry_hub(m, name: "message")
183 today = Time.now.utc.to_date
184 Customer.for_jid(m.from.stripped).then { |customer|
185 sentry_hub.current_scope.set_user(
186 id: customer.customer_id, jid: m.from.stripped.to_s
187 )
188 EMPromise.all([
189 customer, (customer.incr_message_usage if billable_message(m)),
190 REDIS.exists("jmp_usage_notify-#{customer.customer_id}"),
191 customer.stanza_from(m)
192 ])
193 }.then { |(customer, _, already, _)|
194 next if already == 1
195
196 customer.message_usage((today..(today - 30))).then do |usage|
197 next unless usage > 500
198
199 BLATHER.join(CONFIG[:notify_admin], "sgx-jmp")
200 BLATHER.say(
201 CONFIG[:notify_admin],
202 "#{customer.customer_id} has used #{usage} messages since #{today - 30}",
203 :groupchat
204 )
205 REDIS.set("jmp_usage_notify-#{customer.customer_id}", ex: 60 * 60 * 24)
206 end
207 }.catch { |e| panic(e, sentry_hub) }
208end
209
210message :error? do |m|
211 StatsD.increment("message_error")
212
213 puts "MESSAGE ERROR: #{m.inspect}"
214end
215
216class SessionManager
217 def initialize(blather, id_msg, timeout: 5)
218 @blather = blather
219 @sessions = {}
220 @id_msg = id_msg
221 @timeout = timeout
222 end
223
224 def promise_for(stanza)
225 id = "#{stanza.to.stripped}/#{stanza.public_send(@id_msg)}"
226 @sessions.fetch(id) do
227 @sessions[id] = EMPromise.new
228 EM.add_timer(@timeout) do
229 @sessions.delete(id)&.reject(:timeout)
230 end
231 @sessions[id]
232 end
233 end
234
235 def write(stanza)
236 promise = promise_for(stanza)
237 @blather << stanza
238 promise
239 end
240
241 def fulfill(stanza)
242 id = "#{stanza.from.stripped}/#{stanza.public_send(@id_msg)}"
243 if stanza.error?
244 @sessions.delete(id)&.reject(stanza)
245 else
246 @sessions.delete(id)&.fulfill(stanza)
247 end
248 end
249end
250
251IQ_MANAGER = SessionManager.new(self, :id)
252COMMAND_MANAGER = SessionManager.new(self, :sessionid, timeout: 60 * 60)
253web_register_manager = WebRegisterManager.new
254
255disco_info to: Blather::JID.new(CONFIG[:component][:jid]) do |iq|
256 reply = iq.reply
257 reply.identities = [{
258 name: "JMP.chat",
259 type: "sms",
260 category: "gateway"
261 }]
262 reply.features = [
263 "http://jabber.org/protocol/disco#info",
264 "http://jabber.org/protocol/commands"
265 ]
266 form = Blather::Stanza::X.find_or_create(reply.query)
267 form.type = "result"
268 form.fields = [
269 {
270 var: "FORM_TYPE",
271 type: "hidden",
272 value: "http://jabber.org/network/serverinfo"
273 }
274 ] + CONFIG[:xep0157]
275 self << reply
276end
277
278disco_info do |iq|
279 reply = iq.reply
280 reply.identities = [{
281 name: "JMP.chat",
282 type: "sms",
283 category: "client"
284 }]
285 reply.features = [
286 "urn:xmpp:receipts"
287 ]
288 self << reply
289end
290
291disco_items node: "http://jabber.org/protocol/commands" do |iq|
292 StatsD.increment("command_list")
293
294 sentry_hub = new_sentry_hub(iq, name: iq.node)
295 reply = iq.reply
296
297 CommandList.for(iq.from.stripped).then { |list|
298 reply.items = list.map do |item|
299 Blather::Stanza::DiscoItems::Item.new(
300 iq.to,
301 item[:node],
302 item[:name]
303 )
304 end
305 self << reply
306 }.catch { |e| panic(e, sentry_hub) }
307end
308
309iq "/iq/ns:services", ns: "urn:xmpp:extdisco:2" do |iq|
310 StatsD.increment("extdisco")
311
312 reply = iq.reply
313 reply << Nokogiri::XML::Builder.new {
314 services(xmlns: "urn:xmpp:extdisco:2") do
315 service(
316 type: "sip",
317 host: CONFIG[:sip_host]
318 )
319 end
320 }.doc.root
321
322 self << reply
323end
324
325command :execute?, node: "jabber:iq:register", sessionid: nil do |iq|
326 StatsD.increment("command", tags: ["node:#{iq.node}"])
327
328 sentry_hub = new_sentry_hub(iq, name: iq.node)
329 EMPromise.resolve(nil).then {
330 Customer.for_jid(iq.from.stripped)
331 }.catch {
332 sentry_hub.add_breadcrumb(Sentry::Breadcrumb.new(
333 message: "Customer.create"
334 ))
335 Customer.create(iq.from.stripped)
336 }.then { |customer|
337 sentry_hub.current_scope.set_user(
338 id: customer.customer_id,
339 jid: iq.from.stripped.to_s
340 )
341 sentry_hub.add_breadcrumb(Sentry::Breadcrumb.new(
342 message: "Registration.for"
343 ))
344 Registration.for(
345 iq,
346 customer,
347 web_register_manager
348 ).then(&:write)
349 }.catch_only(Blather::Stanza) { |reply|
350 self << reply
351 }.catch { |e| panic(e, sentry_hub) }
352end
353
354def reply_with_note(iq, text, type: :info)
355 reply = iq.reply
356 reply.status = :completed
357 reply.note_type = type
358 reply.note_text = text
359
360 self << reply
361end
362
363# Commands that just pass through to the SGX
364command node: [
365 "number-display",
366 "configure-calls",
367 "record-voicemail-greeting"
368] do |iq|
369 StatsD.increment("command", tags: ["node:#{iq.node}"])
370
371 sentry_hub = new_sentry_hub(iq, name: iq.node)
372 Customer.for_jid(iq.from.stripped).then { |customer|
373 sentry_hub.current_scope.set_user(
374 id: customer.customer_id,
375 jid: iq.from.stripped.to_s
376 )
377
378 customer.stanza_from(iq)
379 }.catch { |e| panic(e, sentry_hub) }
380end
381
382command :execute?, node: "credit cards", sessionid: nil do |iq|
383 StatsD.increment("command", tags: ["node:#{iq.node}"])
384
385 sentry_hub = new_sentry_hub(iq, name: iq.node)
386 reply = iq.reply
387 reply.status = :completed
388
389 Customer.for_jid(iq.from.stripped).then { |customer|
390 oob = OOB.find_or_create(reply.command)
391 oob.url = CONFIG[:credit_card_url].call(
392 reply.to.stripped.to_s.gsub("\\", "%5C"),
393 customer.customer_id
394 )
395 oob.desc = "Manage credits cards and settings"
396
397 reply.note_type = :info
398 reply.note_text = "#{oob.desc}: #{oob.url}"
399
400 self << reply
401 }.catch { |e| panic(e, sentry_hub) }
402end
403
404command :execute?, node: "top up", sessionid: nil do |iq|
405 StatsD.increment("command", tags: ["node:#{iq.node}"])
406
407 sentry_hub = new_sentry_hub(iq, name: iq.node)
408 reply = iq.reply
409 reply.allowed_actions = [:complete]
410
411 Customer.for_jid(iq.from.stripped).then { |customer|
412 BuyAccountCreditForm.for(customer).then do |credit_form|
413 credit_form.add_to_form(reply.form)
414 COMMAND_MANAGER.write(reply).then { |iq2| [customer, credit_form, iq2] }
415 end
416 }.then { |(customer, credit_form, iq2)|
417 iq = iq2 # This allows the catch to use it also
418 Transaction.sale(customer, **credit_form.parse(iq2.form))
419 }.then { |transaction|
420 transaction.insert.then do
421 reply_with_note(iq, "#{transaction} added to your account balance.")
422 end
423 }.catch_only(BuyAccountCreditForm::AmountValidationError) { |e|
424 reply_with_note(iq, e.message, type: :error)
425 }.catch { |e|
426 sentry_hub.capture_exception(e)
427 text = "Failed to buy credit, system said: #{e.message}"
428 reply_with_note(iq, text, type: :error)
429 }.catch { |e| panic(e, sentry_hub) }
430end
431
432command :execute?, node: "alt top up", sessionid: nil do |iq|
433 StatsD.increment("command", tags: ["node:#{iq.node}"])
434
435 sentry_hub = new_sentry_hub(iq, name: iq.node)
436 reply = iq.reply
437 reply.status = :executing
438 reply.allowed_actions = [:complete]
439
440 Customer.for_jid(iq.from.stripped).then { |customer|
441 sentry_hub.current_scope.set_user(
442 id: customer.customer_id,
443 jid: iq.from.stripped.to_s
444 )
445
446 EMPromise.all([AltTopUpForm.for(customer), customer])
447 }.then { |(alt_form, customer)|
448 reply.command << alt_form.form
449
450 COMMAND_MANAGER.write(reply).then do |iq2|
451 AddBitcoinAddress.for(iq2, alt_form, customer).write
452 end
453 }.catch { |e| panic(e, sentry_hub) }
454end
455
456command :execute?, node: "reset sip account", sessionid: nil do |iq|
457 StatsD.increment("command", tags: ["node:#{iq.node}"])
458
459 sentry_hub = new_sentry_hub(iq, name: iq.node)
460 Customer.for_jid(iq.from.stripped).then { |customer|
461 sentry_hub.current_scope.set_user(
462 id: customer.customer_id,
463 jid: iq.from.stripped.to_s
464 )
465 customer.reset_sip_account
466 }.then { |sip_account|
467 reply = iq.reply
468 reply.command << sip_account.form
469 BLATHER << reply
470 }.catch { |e| panic(e, sentry_hub) }
471end
472
473command :execute?, node: "usage", sessionid: nil do |iq|
474 StatsD.increment("command", tags: ["node:#{iq.node}"])
475
476 sentry_hub = new_sentry_hub(iq, name: iq.node)
477 report_for = (Date.today..(Date.today << 1))
478
479 Customer.for_jid(iq.from.stripped).then { |customer|
480 sentry_hub.current_scope.set_user(
481 id: customer.customer_id,
482 jid: iq.from.stripped.to_s
483 )
484
485 customer.usage_report(report_for)
486 }.then { |usage_report|
487 reply = iq.reply
488 reply.status = :completed
489 reply.command << usage_report.form
490 BLATHER << reply
491 }.catch { |e| panic(e, sentry_hub) }
492end
493
494command :execute?, node: "web-register", sessionid: nil do |iq|
495 StatsD.increment("command", tags: ["node:#{iq.node}"])
496
497 sentry_hub = new_sentry_hub(iq, name: iq.node)
498
499 begin
500 jid = iq.form.field("jid")&.value.to_s.strip
501 tel = iq.form.field("tel")&.value.to_s.strip
502 sentry_hub.current_scope.set_user(jid: jid, tel: tel)
503 if iq.from.stripped != CONFIG[:web_register][:from]
504 BLATHER << iq.as_error("forbidden", :auth)
505 elsif jid == "" || tel !~ /\A\+\d+\Z/
506 reply_with_note(iq, "Invalid JID or telephone number.", type: :error)
507 else
508 IQ_MANAGER.write(Blather::Stanza::Iq::Command.new.tap { |cmd|
509 cmd.to = CONFIG[:web_register][:to]
510 cmd.node = "push-register"
511 cmd.form.fields = [var: "to", value: jid]
512 cmd.form.type = "submit"
513 }).then { |result|
514 final_jid = result.form.field("from")&.value.to_s.strip
515 web_register_manager[final_jid] = tel
516 BLATHER << iq.reply.tap { |reply| reply.status = :completed }
517 }.catch { |e| panic(e, sentry_hub) }
518 end
519 rescue StandardError => e
520 sentry_hub.capture_exception(e)
521 end
522end
523
524command sessionid: /./ do |iq|
525 COMMAND_MANAGER.fulfill(iq)
526end
527
528iq type: [:result, :error] do |iq|
529 IQ_MANAGER.fulfill(iq)
530end
531
532iq type: [:get, :set] do |iq|
533 StatsD.increment("unknown_iq")
534
535 self << Blather::StanzaError.new(iq, "feature-not-implemented", :cancel)
536end