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