1# frozen_string_literal: true
2
3require "pg/em"
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/backend_sgx"
28require_relative "lib/bandwidth_tn_order"
29require_relative "lib/btc_sell_prices"
30require_relative "lib/buy_account_credit_form"
31require_relative "lib/customer"
32require_relative "lib/electrum"
33require_relative "lib/em"
34require_relative "lib/payment_methods"
35require_relative "lib/registration"
36require_relative "lib/transaction"
37require_relative "lib/web_register_manager"
38
39ELECTRUM = Electrum.new(**CONFIG[:electrum])
40
41Faraday.default_adapter = :em_synchrony
42BandwidthIris::Client.global_options = {
43 account_id: CONFIG[:creds][:account],
44 username: CONFIG[:creds][:username],
45 password: CONFIG[:creds][:password]
46}
47
48def new_sentry_hub(stanza, name: nil)
49 hub = Sentry.get_current_hub&.new_from_top
50 raise "Sentry.init has not been called" unless hub
51
52 hub.push_scope
53 hub.current_scope.clear_breadcrumbs
54 hub.current_scope.set_transaction_name(name) if name
55 hub.current_scope.set_user(jid: stanza.from.stripped.to_s)
56 hub
57end
58
59# Braintree is not async, so wrap in EM.defer for now
60class AsyncBraintree
61 def initialize(environment:, merchant_id:, public_key:, private_key:, **)
62 @gateway = Braintree::Gateway.new(
63 environment: environment,
64 merchant_id: merchant_id,
65 public_key: public_key,
66 private_key: private_key
67 )
68 end
69
70 def respond_to_missing?(m, *)
71 @gateway.respond_to?(m)
72 end
73
74 def method_missing(m, *args)
75 return super unless respond_to_missing?(m, *args)
76
77 EM.promise_defer(klass: PromiseChain) do
78 @gateway.public_send(m, *args)
79 end
80 end
81
82 class PromiseChain < EMPromise
83 def respond_to_missing?(*)
84 false # We don't actually know what we respond to...
85 end
86
87 def method_missing(m, *args)
88 return super if respond_to_missing?(m, *args)
89 self.then { |o| o.public_send(m, *args) }
90 end
91 end
92end
93
94BRAINTREE = AsyncBraintree.new(**CONFIG[:braintree])
95
96def panic(e, hub=nil)
97 m = e.respond_to?(:message) ? e.message : e
98 warn "Error raised during event loop: #{e.class}: #{m}"
99 warn e.backtrace if e.respond_to?(:backtrace)
100 if e.is_a?(::Exception)
101 (hub || Sentry).capture_exception(e, hint: { background: false })
102 else
103 (hub || Sentry).capture_message(e, hint: { background: false })
104 end
105 exit 1
106end
107
108EM.error_handler(&method(:panic))
109
110when_ready do
111 BLATHER = self
112 REDIS = EM::Hiredis.connect
113 BTC_SELL_PRICES = BTCSellPrices.new(REDIS, CONFIG[:oxr_app_id])
114 DB = PG::EM::Client.new(dbname: "jmp")
115 DB.type_map_for_results = PG::BasicTypeMapForResults.new(DB)
116 DB.type_map_for_queries = PG::BasicTypeMapForQueries.new(DB)
117
118 EM.add_periodic_timer(3600) do
119 ping = Blather::Stanza::Iq::Ping.new(:get, CONFIG[:server][:host])
120 ping.from = CONFIG[:component][:jid]
121 self << ping
122 end
123end
124
125# workqueue_count MUST be 0 or else Blather uses threads!
126setup(
127 CONFIG[:component][:jid],
128 CONFIG[:component][:secret],
129 CONFIG[:server][:host],
130 CONFIG[:server][:port],
131 nil,
132 nil,
133 workqueue_count: 0
134)
135
136message to: /\Aaccount@/ do |m|
137 self << m.reply.tap do |out|
138 out.body = "This bot is deprecated. Please talk to xmpp:cheogram.com"
139 end
140end
141
142message to: /\Acustomer_/, from: /@#{CONFIG[:sgx]}(\/|\Z)/ do |m|
143 sentry_hub = new_sentry_hub(iq, name: iq.node)
144 Customer.for_customer_id(
145 m.to.node.delete_prefix("customer_")
146 ).then { |customer|
147 sentry_hub.current_scope.set_user(
148 id: customer.customer_id,
149 jid: iq.from.stripped.to_s
150 )
151 customer.stanza_to(m)
152 }.catch { |e| panic(e, sentry_hub) }
153end
154
155message do |m|
156 sentry_hub = new_sentry_hub(iq, name: iq.node)
157 Customer.for_jid(m.from.stripped).then { |customer|
158 sentry_hub.current_scope.set_user(
159 id: customer.customer_id,
160 jid: iq.from.stripped.to_s
161 )
162 today = Time.now.utc.to_date
163 EMPromise.all([
164 REDIS.zremrangebylex(
165 "jmp_customer_outbound_messages-#{customer.customer_id}",
166 "-",
167 # Store message counts per day for 1 year
168 "[#{(today << 12).strftime('%Y%m%d')}"
169 ),
170 REDIS.zincrby(
171 "jmp_customer_outbound_messages-#{customer.customer_id}",
172 1,
173 today.strftime("%Y%m%d")
174 ),
175 customer.stanza_from(m)
176 ])
177 }.catch { |e| panic(e, sentry_hub) }
178end
179
180message :error? do |m|
181 puts "MESSAGE ERROR: #{m.inspect}"
182end
183
184class SessionManager
185 def initialize(blather, id_msg, timeout: 5)
186 @blather = blather
187 @sessions = {}
188 @id_msg = id_msg
189 @timeout = timeout
190 end
191
192 def promise_for(stanza)
193 id = "#{stanza.to.stripped}/#{stanza.public_send(@id_msg)}"
194 @sessions.fetch(id) do
195 @sessions[id] = EMPromise.new
196 EM.add_timer(@timeout) do
197 @sessions.delete(id)&.reject(:timeout)
198 end
199 @sessions[id]
200 end
201 end
202
203 def write(stanza)
204 promise = promise_for(stanza)
205 @blather << stanza
206 promise
207 end
208
209 def fulfill(stanza)
210 id = "#{stanza.from.stripped}/#{stanza.public_send(@id_msg)}"
211 if stanza.error?
212 @sessions.delete(id)&.reject(stanza)
213 else
214 @sessions.delete(id)&.fulfill(stanza)
215 end
216 end
217end
218
219IQ_MANAGER = SessionManager.new(self, :id)
220COMMAND_MANAGER = SessionManager.new(self, :sessionid, timeout: 60 * 60)
221web_register_manager = WebRegisterManager.new
222
223disco_info to: Blather::JID.new(CONFIG[:component][:jid]) do |iq|
224 reply = iq.reply
225 reply.identities = [{
226 name: "JMP.chat",
227 type: "sms",
228 category: "gateway"
229 }]
230 form = Blather::Stanza::X.find_or_create(reply.query)
231 form.type = "result"
232 form.fields = [
233 {
234 var: "FORM_TYPE",
235 type: "hidden",
236 value: "http://jabber.org/network/serverinfo"
237 }
238 ] + CONFIG[:xep0157]
239 self << reply
240end
241
242disco_items node: "http://jabber.org/protocol/commands" do |iq|
243 reply = iq.reply
244 reply.items = [
245 # TODO: don't show this item if no braintree methods available
246 # TODO: don't show this item if no plan for this customer
247 Blather::Stanza::DiscoItems::Item.new(
248 iq.to,
249 "buy-credit",
250 "Buy account credit"
251 ),
252 Blather::Stanza::DiscoItems::Item.new(
253 iq.to,
254 "jabber:iq:register",
255 "Register"
256 )
257 ]
258 self << reply
259end
260
261command :execute?, node: "jabber:iq:register", sessionid: nil do |iq|
262 sentry_hub = new_sentry_hub(iq, name: iq.node)
263 EMPromise.resolve(nil).then {
264 Customer.for_jid(iq.from.stripped)
265 }.catch {
266 sentry_hub.add_breadcrumb(Sentry::Breadcrumb.new(
267 message: "Customer.create"
268 ))
269 Customer.create(iq.from.stripped)
270 }.then { |customer|
271 sentry_hub.current_scope.set_user(
272 id: customer.customer_id,
273 jid: iq.from.stripped.to_s
274 )
275 sentry_hub.add_breadcrumb(Sentry::Breadcrumb.new(
276 message: "Registration.for"
277 ))
278 Registration.for(
279 iq,
280 customer,
281 web_register_manager
282 ).then(&:write)
283 }.catch { |e| panic(e, sentry_hub) }
284end
285
286def reply_with_note(iq, text, type: :info)
287 reply = iq.reply
288 reply.status = :completed
289 reply.note_type = type
290 reply.note_text = text
291
292 self << reply
293end
294
295command :execute?, node: "buy-credit", sessionid: nil do |iq|
296 sentry_hub = new_sentry_hub(iq, name: iq.node)
297 reply = iq.reply
298 reply.allowed_actions = [:complete]
299
300 Customer.for_jid(iq.from.stripped).then { |customer|
301 BuyAccountCreditForm.for(customer).then do |credit_form|
302 credit_form.add_to_form(reply.form)
303 COMMAND_MANAGER.write(reply).then { |iq2| [customer, credit_form, iq2] }
304 end
305 }.then { |(customer, credit_form, iq2)|
306 iq = iq2 # This allows the catch to use it also
307 Transaction.sale(customer, **credit_form.parse(iq2.form))
308 }.then { |transaction|
309 transaction.insert.then { transaction.amount }
310 }.then { |amount|
311 reply_with_note(iq, "$#{'%.2f' % amount} added to your account balance.")
312 }.catch { |e|
313 sentry_hub.capture_exception(e)
314 text = "Failed to buy credit, system said: #{e.message}"
315 reply_with_note(iq, text, type: :error)
316 }.catch { |e| panic(e, sentry_hub) }
317end
318
319command :execute?, node: "web-register", sessionid: nil do |iq|
320 sentry_hub = new_sentry_hub(iq, name: iq.node)
321
322 begin
323 jid = iq.form.field("jid")&.value.to_s.strip
324 tel = iq.form.field("tel")&.value.to_s.strip
325 hub.current_scope.set_user(jid: jid, tel: tel)
326 if iq.from.stripped != CONFIG[:web_register][:from]
327 BLATHER << iq.as_error("forbidden", :auth)
328 elsif jid == "" || tel !~ /\A\+\d+\Z/
329 reply_with_note(iq, "Invalid JID or telephone number.", type: :error)
330 else
331 IQ_MANAGER.write(Blather::Stanza::Iq::Command.new.tap { |cmd|
332 cmd.to = CONFIG[:web_register][:to]
333 cmd.from = CONFIG[:component][:jid]
334 cmd.node = "push-register"
335 cmd.form.fields = [var: "to", value: jid]
336 cmd.form.type = "submit"
337 }).then { |result|
338 final_jid = result.form.field("from")&.value.to_s.strip
339 web_register_manager[final_jid] = tel
340 BLATHER << iq.reply.tap { |reply| reply.status = :completed }
341 }.catch { |e| panic(e, sentry_hub) }
342 end
343 rescue StandardError => e
344 sentry_hub.capture_exception(e)
345 end
346end
347
348command sessionid: /./ do |iq|
349 COMMAND_MANAGER.fulfill(iq)
350end
351
352iq :result? do |iq|
353 IQ_MANAGER.fulfill(iq)
354end
355
356iq :error? do |iq|
357 IQ_MANAGER.fulfill(iq)
358end