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