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