1#!/usr/bin/ruby
2# frozen_string_literal: true
3
4require "bigdecimal"
5require "delegate"
6require "em_promise"
7require "set"
8
9SCHEMA = "{
10 xmpp: { jid: Text, password: Text },
11 component: { jid: Text },
12 keepgo: Optional { access_token: Text, api_key: Text },
13 sims: {
14 annual: { CAD: Natural, USD: Natural },
15 per_gb: { CAD: Natural, USD: Natural }
16 },
17 plans: List {
18 currency: < CAD | USD >,
19 messages: < limited: { included: Natural, price: Natural } | unlimited >,
20 minutes: < limited: { included: Natural, price: Natural } | unlimited >,
21 monthly_price : Natural,
22 name : Text,
23 allow_register: Bool,
24 subaccount_discount: Natural
25 },
26 braintree: {
27 environment : Text,
28 merchant_accounts : { CAD : Text, USD : Text },
29 merchant_id : Text,
30 private_key : Text,
31 public_key : Text
32 }
33}"
34
35class JobCustomer < SimpleDelegator
36 def initialize(customer, billing_customer: nil)
37 @billing_customer = billing_customer
38 super(customer)
39 end
40
41 def auto_top_up_amount
42 @billing_customer&.auto_top_up_amount || super
43 end
44
45 def billing_customer(*)
46 if @billing_customer
47 return EMPromise.resolve(self.class.new(@billing_customer))
48 end
49
50 super.then(&self.class.method(:new))
51 end
52
53 def with(balance:)
54 self.class.new(
55 __getobj__.with(balance: balance),
56 billing_customer: @billing_customer&.with(balance: balance)
57 )
58 end
59
60 def stanza_to(stanza)
61 stanza = stanza.dup
62 stanza.from = nil # It's a client connection, use default
63 stanza.to = Blather::JID.new(
64 "customer_#{customer_id}", CONFIG[:component][:jid]
65 ).with(resource: stanza.to&.resource)
66 block_given? ? yield(stanza) : (BlatherNotify << stanza)
67 end
68end
69
70module SimAction
71 attr_accessor :customer
72
73 def initialize(sim, customer: nil)
74 @sim = sim
75 @customer = customer
76 end
77
78 def iccid
79 @sim.iccid
80 end
81
82 def customer_id
83 customer.customer_id
84 end
85
86 def refill_price
87 SIMPricing.five_gb_refill_price(customer.currency, CONFIG)
88 end
89
90 def refill_and_bill(data, price, note)
91 SIM_REPO.refill(@sim, amount_mb: data).then { |keepgo_tx|
92 raise "SIM refill failed: #{iccid}" unless keepgo_tx["ack"] == "success"
93
94 Transaction.new(
95 customer_id: customer_id,
96 transaction_id: keepgo_tx["transaction_id"],
97 amount: -price, note: note
98 ).insert
99 }.then do
100 LOG.info "Refilled #{customer.customer_id} #{iccid}"
101 end
102 end
103
104 def monthly_limit
105 REDIS.get(
106 "jmp_customer_monthly_data_limit-#{customer_id}"
107 ).then do |limit|
108 BigDecimal(limit || refill_price)
109 end
110 end
111
112 def amount_spent
113 promise = DB.query_defer(<<~SQL, [customer_id])
114 SELECT COALESCE(SUM(amount), 0) AS a FROM transactions WHERE
115 customer_id=$1 AND transaction_id LIKE 'AB_59576_%' AND
116 created_at >= DATE_TRUNC('month', LOCALTIMESTAMP)
117 SQL
118 promise.then { |rows| -rows.first["a"] }
119 end
120end
121
122class SimTopUp
123 include SimAction
124
125 def low_balance
126 LowBalance.for(customer, refill_price).then(&:notify!).then do |result|
127 @customer = customer.with(balance: customer.balance + result)
128 next call if result.positive?
129
130 LOG.info "Low balance #{customer.customer_id} #{iccid}"
131 end
132 end
133
134 def call
135 EMPromise.all([amount_spent, monthly_limit]).then do |(spent, limit)|
136 if monthly_limit_allows?(spent, limit)
137 next low_balance if customer.balance < refill_price
138
139 refill_and_bill(5120, refill_price, "5GB Data Topup for #{iccid}")
140 else
141 warn
142 end
143 end
144 end
145
146 def warn
147 SimWarn.new(@sim, customer: customer).notify
148 end
149
150 def monthly_limit_allows?(spent, limit)
151 spent + refill_price <= limit
152 end
153end
154
155class SimWarn
156 include SimAction
157
158 def notify
159 ExpiringLock.new("jmp_customer_sim_warn-#{customer.customer_id}").with do
160 m = Blather::Stanza::Message.new
161 m.body = "Your SIM #{iccid} only has " \
162 "#{(@sim.remaining_usage_kb / 1024).to_i} MB left"
163 customer.stanza_to(m)
164 LOG.info "Data warning #{customer.customer_id} #{@sim.iccid}"
165 end
166 end
167
168 def call
169 EMPromise.all([amount_spent, monthly_limit]).then do |(spent, limit)|
170 next unless notify?(spent, limit)
171
172 notify
173 end
174 end
175
176 def notify?(spent, limit)
177 spent + refill_price > limit || low_balance_and_not_auto_top_up
178 end
179
180 def low_balance_and_not_auto_top_up
181 customer.balance < refill_price && !customer.auto_top_up_amount&.positive?
182 end
183end
184
185class SimAnnual
186 include SimAction
187
188 def notify
189 ExpiringLock.new("jmp_customer_sim_annual-#{customer.customer_id}").with do
190 m = Blather::Stanza::Message.new
191 m.body = "Your SIM #{iccid} only has #{@sim.remaining_days} days left"
192 customer.stanza_to(m)
193 LOG.info "Annual warning #{customer.customer_id} #{@sim.iccid}"
194 end
195 end
196
197 def annual_price
198 SIMPricing.annual_price(customer.currency, CONFIG)
199 end
200
201 def call
202 if customer.balance >= annual_price
203 refill_and_bill(1024, annual_price, "Annual fee for #{iccid}")
204 else
205 low_balance
206 end
207 end
208
209 def low_balance
210 LowBalance.for(customer, annual_price).then(&:notify!).then do |result|
211 @customer = customer.with(balance: customer.balance + result)
212 next call if result.positive?
213
214 notify
215 end
216 end
217end
218
219def fetch_customers(cids, customer_repo: CUSTOMER_REPO)
220 # This is gross N+1 against the DB, but also does a buch of Redis work
221 # We expect the set to be very small for the forseeable future,
222 # hundreds at most
223 EMPromise.all(
224 Set.new(cids).to_a.compact.map { |id|
225 customer_repo.find(id).catch_only(CustomerRepo::NotFound) { nil }
226 }
227 ).then do |customers|
228 Hash[customers.compact.map { |c| [c.customer_id, JobCustomer.new(c)] }]
229 end
230end
231
232SIM_QUERY = "SELECT iccid, customer_id FROM sims WHERE iccid = ANY ($1)"
233def load_customers!(sims, db: DB, customer_repo: CUSTOMER_REPO)
234 db.query_defer(SIM_QUERY, [sims.keys]).then { |rows|
235 load_customer_rows!(sims, rows, customer_repo)
236 }
237end
238
239def load_customer_rows!(sims, rows, customer_repo)
240 fetch_customers(
241 rows.map { |row| row["customer_id"] },
242 customer_repo: customer_repo
243 ).then do |customers|
244 rows.each do |row|
245 sims[row["iccid"]]&.customer = customers[row["customer_id"]]
246 end
247
248 sims
249 end
250end
251
252def decide_sim_actions(sims)
253 sims.each_with_object({}) { |sim, h|
254 if sim.remaining_days < 31
255 h[sim.iccid] = SimAnnual.new(sim)
256 elsif sim.remaining_usage_kb < 100000
257 h[sim.iccid] = SimTopUp.new(sim)
258 elsif sim.remaining_usage_kb < 250000
259 h[sim.iccid] = SimWarn.new(sim)
260 end
261 }.compact
262end
263
264def report_sim_job_error(e)
265 LOG.error e
266
267 case e
268 when Exception then Sentry.capture_exception(e)
269 when EM::HttpClient
270 LOG.error e.response_header
271 Sentry.capture_message("HTTP Error: #{e.response_header.http_status} " \
272 "@ #{e.last_effective_url}")
273 else
274 Sentry.capture_message(e.to_s)
275 end
276end
277
278def process_sim_actions(actions, customer_repo: CUSTOMER_REPO)
279 actions.reduce(EMPromise.resolve(nil)) do |promise, action|
280 promise.then { process_sim_action(action, customer_repo) }
281 end
282end
283
284def process_sim_action(action, customer_repo)
285 EMPromise.resolve(nil).then {
286 reload_action_customer(action, customer_repo).then { |customer|
287 action.customer = customer
288 action.call
289 }
290 }.catch { |e|
291 report_sim_job_error(e)
292 nil
293 }
294end
295
296def reload_action_customer(action, customer_repo)
297 customer_repo.find(action.customer.customer_id).then do |customer|
298 customer.billing_customer(customer_repo).then do |billing_customer|
299 JobCustomer.new(
300 customer.with(balance: billing_customer.balance),
301 billing_customer: billing_customer
302 )
303 end
304 end
305end
306
307if $0 == __FILE__
308 require "pg/em/connection_pool"
309 require "eventmachine"
310 require "em-hiredis"
311 require "dhall"
312 require "ougai"
313 require "sentry-ruby"
314
315 $stdout.sync = true
316 LOG = Ougai::Logger.new($stdout)
317 LOG.level = ENV.fetch("LOG_LEVEL", "info")
318 LOG.formatter = Ougai::Formatters::Readable.new(
319 nil,
320 nil,
321 plain: !$stdout.isatty
322 )
323
324 def log
325 LOG
326 end
327
328 Sentry.init do |config|
329 config.background_worker_threads = 0
330 end
331
332 raise "Need a Dhall config" unless ARGV[0]
333
334 CONFIG = Dhall::Coder
335 .new(safe: Dhall::Coder::JSON_LIKE + [Symbol, Proc])
336 .load("#{ARGV.first} : #{SCHEMA}", transform_keys: :to_sym)
337
338 CONFIG[:keep_area_codes_in] = {}
339 CONFIG[:creds] = {}
340
341 require_relative "../lib/async_braintree"
342 require_relative "../lib/blather_notify"
343 require_relative "../lib/customer_repo"
344 require_relative "../lib/expiring_lock"
345 require_relative "../lib/low_balance"
346 require_relative "../lib/postgres"
347 require_relative "../lib/sim_repo"
348 require_relative "../lib/sim_pricing"
349 require_relative "../lib/transaction"
350
351 BRAINTREE = AsyncBraintree.new(**CONFIG[:braintree])
352 CUSTOMER_REPO = CustomerRepo.new
353 SIM_REPO = SIMRepo.new
354
355 EM.run do
356 REDIS = EM::Hiredis.connect
357 DB = Postgres.connect(dbname: "jmp")
358
359 BlatherNotify.start(
360 CONFIG[:xmpp][:jid],
361 CONFIG[:xmpp][:password]
362 ).then { SIM_REPO.all }.then { |sims|
363 load_customers!(decide_sim_actions(sims))
364 }.then { |items|
365 items = items.values.select { |item| item.customer&.currency }
366 process_sim_actions(items)
367 }.catch(&method(:report_sim_job_error)).then { EM.stop }
368 end
369end