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