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/) {}
174
175def billable_message(m)
176 !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,
190 (customer.incr_message_usage if billable_message(m)),
191 REDIS.exists("jmp_usage_notify-#{customer.customer_id}"),
192 customer.stanza_from(m)
193 ])
194 }.then { |(customer, _, already, _)|
195 next if already == 1
196
197 customer.message_usage((today..(today - 30))).then do |usage|
198 next unless usage > 500
199
200 BLATHER.join(CONFIG[:notify_admin], "sgx-jmp")
201 BLATHER.say(
202 CONFIG[:notify_admin],
203 "#{customer.customer_id} has used #{usage} messages since #{today - 30}"
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 form = Blather::Stanza::X.find_or_create(reply.query)
263 form.type = "result"
264 form.fields = [
265 {
266 var: "FORM_TYPE",
267 type: "hidden",
268 value: "http://jabber.org/network/serverinfo"
269 }
270 ] + CONFIG[:xep0157]
271 self << reply
272end
273
274disco_items node: "http://jabber.org/protocol/commands" do |iq|
275 StatsD.increment("command_list")
276
277 sentry_hub = new_sentry_hub(iq, name: iq.node)
278 reply = iq.reply
279
280 CommandList.for(iq.from.stripped).then { |list|
281 reply.items = list.map do |item|
282 Blather::Stanza::DiscoItems::Item.new(
283 iq.to,
284 item[:node],
285 item[:name]
286 )
287 end
288 self << reply
289 }.catch { |e| panic(e, sentry_hub) }
290end
291
292iq "/iq/ns:services", ns: "urn:xmpp:extdisco:2" do |iq|
293 StatsD.increment("extdisco")
294
295 reply = iq.reply
296 reply << Nokogiri::XML::Builder.new {
297 services(xmlns: "urn:xmpp:extdisco:2") do
298 service(
299 type: "sip",
300 host: CONFIG[:sip_host]
301 )
302 end
303 }.doc.root
304
305 self << reply
306end
307
308command :execute?, node: "jabber:iq:register", sessionid: nil do |iq|
309 StatsD.increment("command", tags: ["node:#{iq.node}"])
310
311 sentry_hub = new_sentry_hub(iq, name: iq.node)
312 EMPromise.resolve(nil).then {
313 Customer.for_jid(iq.from.stripped)
314 }.catch {
315 sentry_hub.add_breadcrumb(Sentry::Breadcrumb.new(
316 message: "Customer.create"
317 ))
318 Customer.create(iq.from.stripped)
319 }.then { |customer|
320 sentry_hub.current_scope.set_user(
321 id: customer.customer_id,
322 jid: iq.from.stripped.to_s
323 )
324 sentry_hub.add_breadcrumb(Sentry::Breadcrumb.new(
325 message: "Registration.for"
326 ))
327 Registration.for(
328 iq,
329 customer,
330 web_register_manager
331 ).then(&:write)
332 }.catch_only(Blather::Stanza) { |reply|
333 self << reply
334 }.catch { |e| panic(e, sentry_hub) }
335end
336
337def reply_with_note(iq, text, type: :info)
338 reply = iq.reply
339 reply.status = :completed
340 reply.note_type = type
341 reply.note_text = text
342
343 self << reply
344end
345
346# Commands that just pass through to the SGX
347command node: [
348 "number-display",
349 "configure-calls",
350 "record-voicemail-greeting"
351] do |iq|
352 StatsD.increment("command", tags: ["node:#{iq.node}"])
353
354 sentry_hub = new_sentry_hub(iq, name: iq.node)
355 Customer.for_jid(iq.from.stripped).then { |customer|
356 sentry_hub.current_scope.set_user(
357 id: customer.customer_id,
358 jid: iq.from.stripped.to_s
359 )
360
361 customer.stanza_from(iq)
362 }.catch { |e| panic(e, sentry_hub) }
363end
364
365command :execute?, node: "credit cards", sessionid: nil do |iq|
366 StatsD.increment("command", tags: ["node:#{iq.node}"])
367
368 sentry_hub = new_sentry_hub(iq, name: iq.node)
369 reply = iq.reply
370 reply.status = :completed
371
372 Customer.for_jid(iq.from.stripped).then { |customer|
373 oob = OOB.find_or_create(reply.command)
374 oob.url = CONFIG[:credit_card_url].call(
375 reply.to.stripped.to_s.gsub("\\", "%5C"),
376 customer.customer_id
377 )
378 oob.desc = "Manage credits cards and settings"
379
380 reply.note_type = :info
381 reply.note_text = "#{oob.desc}: #{oob.url}"
382
383 self << reply
384 }.catch { |e| panic(e, sentry_hub) }
385end
386
387command :execute?, node: "top up", sessionid: nil do |iq|
388 StatsD.increment("command", tags: ["node:#{iq.node}"])
389
390 sentry_hub = new_sentry_hub(iq, name: iq.node)
391 reply = iq.reply
392 reply.allowed_actions = [:complete]
393
394 Customer.for_jid(iq.from.stripped).then { |customer|
395 BuyAccountCreditForm.for(customer).then do |credit_form|
396 credit_form.add_to_form(reply.form)
397 COMMAND_MANAGER.write(reply).then { |iq2| [customer, credit_form, iq2] }
398 end
399 }.then { |(customer, credit_form, iq2)|
400 iq = iq2 # This allows the catch to use it also
401 Transaction.sale(customer, **credit_form.parse(iq2.form))
402 }.then { |transaction|
403 transaction.insert.then do
404 reply_with_note(iq, "#{transaction} added to your account balance.")
405 end
406 }.catch_only(BuyAccountCreditForm::AmountValidationError) { |e|
407 reply_with_note(iq, e.message, type: :error)
408 }.catch { |e|
409 sentry_hub.capture_exception(e)
410 text = "Failed to buy credit, system said: #{e.message}"
411 reply_with_note(iq, text, type: :error)
412 }.catch { |e| panic(e, sentry_hub) }
413end
414
415command :execute?, node: "alt top up", sessionid: nil do |iq|
416 StatsD.increment("command", tags: ["node:#{iq.node}"])
417
418 sentry_hub = new_sentry_hub(iq, name: iq.node)
419 reply = iq.reply
420 reply.status = :executing
421 reply.allowed_actions = [:complete]
422
423 Customer.for_jid(iq.from.stripped).then { |customer|
424 sentry_hub.current_scope.set_user(
425 id: customer.customer_id,
426 jid: iq.from.stripped.to_s
427 )
428
429 EMPromise.all([AltTopUpForm.for(customer), customer])
430 }.then { |(alt_form, customer)|
431 reply.command << alt_form.form
432
433 COMMAND_MANAGER.write(reply).then do |iq2|
434 AddBitcoinAddress.for(iq2, alt_form, customer).write
435 end
436 }.catch { |e| panic(e, sentry_hub) }
437end
438
439command :execute?, node: "reset sip account", sessionid: nil do |iq|
440 StatsD.increment("command", tags: ["node:#{iq.node}"])
441
442 sentry_hub = new_sentry_hub(iq, name: iq.node)
443 Customer.for_jid(iq.from.stripped).then { |customer|
444 sentry_hub.current_scope.set_user(
445 id: customer.customer_id,
446 jid: iq.from.stripped.to_s
447 )
448 customer.reset_sip_account
449 }.then { |sip_account|
450 reply = iq.reply
451 reply.command << sip_account.form
452 BLATHER << reply
453 }.catch { |e| panic(e, sentry_hub) }
454end
455
456command :execute?, node: "usage", sessionid: nil do |iq|
457 StatsD.increment("command", tags: ["node:#{iq.node}"])
458
459 sentry_hub = new_sentry_hub(iq, name: iq.node)
460 report_for = (Date.today..(Date.today << 1))
461
462 Customer.for_jid(iq.from.stripped).then { |customer|
463 sentry_hub.current_scope.set_user(
464 id: customer.customer_id,
465 jid: iq.from.stripped.to_s
466 )
467
468 customer.usage_report(report_for)
469 }.then { |usage_report|
470 reply = iq.reply
471 reply.status = :completed
472 reply.command << usage_report.form
473 BLATHER << reply
474 }.catch { |e| panic(e, sentry_hub) }
475end
476
477command :execute?, node: "web-register", sessionid: nil do |iq|
478 StatsD.increment("command", tags: ["node:#{iq.node}"])
479
480 sentry_hub = new_sentry_hub(iq, name: iq.node)
481
482 begin
483 jid = iq.form.field("jid")&.value.to_s.strip
484 tel = iq.form.field("tel")&.value.to_s.strip
485 hub.current_scope.set_user(jid: jid, tel: tel)
486 if iq.from.stripped != CONFIG[:web_register][:from]
487 BLATHER << iq.as_error("forbidden", :auth)
488 elsif jid == "" || tel !~ /\A\+\d+\Z/
489 reply_with_note(iq, "Invalid JID or telephone number.", type: :error)
490 else
491 IQ_MANAGER.write(Blather::Stanza::Iq::Command.new.tap { |cmd|
492 cmd.to = CONFIG[:web_register][:to]
493 cmd.node = "push-register"
494 cmd.form.fields = [var: "to", value: jid]
495 cmd.form.type = "submit"
496 }).then { |result|
497 final_jid = result.form.field("from")&.value.to_s.strip
498 web_register_manager[final_jid] = tel
499 BLATHER << iq.reply.tap { |reply| reply.status = :completed }
500 }.catch { |e| panic(e, sentry_hub) }
501 end
502 rescue StandardError => e
503 sentry_hub.capture_exception(e)
504 end
505end
506
507command sessionid: /./ do |iq|
508 COMMAND_MANAGER.fulfill(iq)
509end
510
511iq type: [:result, :error] do |iq|
512 IQ_MANAGER.fulfill(iq)
513end
514
515iq type: [:get, :set] do |iq|
516 StatsD.increment("unknown_iq")
517
518 self << Blather::StanzaError.new(iq, "feature-not-implemented", :cancel)
519end