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