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