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