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