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