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