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