1"""Chapter 33 (case study: notification system, OTP as the running example) deep-dive code.
21. One queue per channel/priority: a marketing blast must not delay OTPs.
32. Retries, provider failover and de-duplication.
43. User preferences: opt-outs, quiet hours, frequency caps, transactional vs marketing.
54. OTPs: expiry, attempts, resend limits, hashed storage.
6Run: python3 ch33_notifications.py"""
7import hashlib
8import hmac
9import secrets
10
11
12# ---- 1. queues ---------------------------------------------------------------------
13
14def wait_in_shared_queue(backlog_ahead, workers_per_s):
15 """A FIFO shared by everything: an OTP waits behind the whole backlog."""
16 return backlog_ahead / workers_per_s
17
18
19def wait_in_own_queue(otp_backlog, otp_workers_per_s):
20 return otp_backlog / otp_workers_per_s
21
22
23# ---- 2. retries, failover, de-duplication -------------------------------------------
24
25class Provider:
26 def __init__(self, name, fail_first=0):
27 self.name, self.fail_first, self.sent = name, fail_first, []
28
29 def send(self, to, body, idem_key):
30 if self.fail_first > 0:
31 self.fail_first -= 1
32 raise TimeoutError(self.name)
33 self.sent.append((to, body, idem_key))
34 return f"{self.name}-msg-{len(self.sent)}"
35
36
37class Sender:
38 def __init__(self, providers, max_attempts=4):
39 self.providers, self.max_attempts = providers, max_attempts
40 self.done = {} # idempotency key -> provider message id
41
42 def deliver(self, notif):
43 key = notif["idem_key"]
44 if key in self.done: # duplicate event from the queue
45 return self.done[key], "duplicate-skipped"
46 for attempt in range(self.max_attempts):
47 provider = self.providers[attempt % len(self.providers)] # fail over on each retry
48 try:
49 mid = provider.send(notif["to"], notif["body"], key)
50 self.done[key] = mid
51 return mid, f"sent via {provider.name} on attempt {attempt + 1}"
52 except TimeoutError:
53 continue # real code: backoff with jitter here
54 return None, "dead-lettered"
55
56
57# ---- 3. preferences ----------------------------------------------------------------
58
59def decide(notif, prefs, local_hour, sent_today):
60 """Returns the channels to use, or [] with a reason."""
61 if notif["category"] == "security": # OTPs, password resets: always, on the chosen channel
62 return [notif["channel"]], "security: preferences don't block it"
63 channels = [c for c in notif["channels"] if prefs["opt_in"].get((notif["category"], c), False)]
64 if not channels:
65 return [], "opted out"
66 if notif["category"] == "marketing":
67 if 22 <= local_hour or local_hour < 8:
68 return [], "quiet hours: schedule for 08:00 local"
69 if sent_today >= prefs["daily_cap"]:
70 return [], "frequency cap reached"
71 return channels, "ok"
72
73
74# ---- 4. OTPs ----------------------------------------------------------------------
75
76class OtpService:
77 def __init__(self, ttl=300, max_attempts=5, max_sends_per_hour=3, secret=b"server-pepper"):
78 self.ttl, self.max_attempts, self.max_sends = ttl, max_attempts, max_sends_per_hour
79 self.secret, self.store, self.sends = secret, {}, {}
80
81 def _hash(self, phone, code):
82 return hmac.new(self.secret, f"{phone}:{code}".encode(), hashlib.sha256).hexdigest()
83
84 def request(self, phone, now, rnd=secrets.SystemRandom()):
85 recent = [t for t in self.sends.get(phone, []) if now - t < 3600]
86 if len(recent) >= self.max_sends:
87 return None, "too many OTPs: try later"
88 code = f"{rnd.randrange(1_000_000):06d}" # cryptographic randomness by default
89 self.store[phone] = {"hash": self._hash(phone, code), "expires": now + self.ttl, "attempts": 0}
90 self.sends[phone] = recent + [now]
91 return code, "sent"
92
93 def verify(self, phone, code, now):
94 rec = self.store.get(phone)
95 if not rec:
96 return False, "no OTP"
97 if now > rec["expires"]:
98 return False, "expired"
99 if rec["attempts"] >= self.max_attempts:
100 return False, "locked: too many attempts"
101 rec["attempts"] += 1
102 if hmac.compare_digest(rec["hash"], self._hash(phone, code)):
103 del self.store[phone] # single use
104 return True, "ok"
105 return False, "wrong code"
106
107
108def should_still_send(created_at, now, ttl):
109 """A worker picking up an OTP job late should drop it if the code has expired."""
110 return now - created_at < ttl
111
112
113# ---- checks ------------------------------------------------------------------------
114
115def run_checks():
116 # 1. a 2-million-email sale campaign is queued just before Asha asks for an OTP
117 assert wait_in_shared_queue(2_000_000, 5_000) == 400 # 6 minutes 40 s: the OTP is useless
118 assert wait_in_own_queue(30, 300) == 0.1 # own queue: a tenth of a second
119
120 # 2. provider A times out twice; retries fail over to B; the queue redelivers the event
121 a, b = Provider("sms-a", fail_first=2), Provider("sms-b")
122 s = Sender([a, b])
123 otp = {"idem_key": "otp:asha:req-77", "to": "+91...", "body": "Your code is 482913"}
124 mid, how = s.deliver(otp)
125 assert how == "sent via sms-b on attempt 2" and mid == "sms-b-msg-1"
126 mid2, how2 = s.deliver(otp) # at-least-once delivery from the queue
127 assert how2 == "duplicate-skipped" and len(b.sent) == 1 # one SMS, not two
128 dead = Sender([Provider("x", fail_first=9), Provider("y", fail_first=9)])
129 assert dead.deliver({"idem_key": "k", "to": "t", "body": "b"})[1] == "dead-lettered"
130
131 # 3. preferences
132 prefs = {"opt_in": {("order", "push"): True, ("order", "email"): True,
133 ("marketing", "email"): True}, "daily_cap": 2}
134 order = {"category": "order", "channels": ["push", "sms", "email"]}
135 promo = {"category": "marketing", "channels": ["push", "email"]}
136 login = {"category": "security", "channel": "sms"}
137 assert decide(order, prefs, 14, 0) == (["push", "email"], "ok")
138 assert decide(promo, prefs, 23, 0)[1].startswith("quiet hours")
139 assert decide(promo, prefs, 14, 2)[1] == "frequency cap reached"
140 assert decide(promo, prefs, 14, 0) == (["email"], "ok")
141 assert decide(login, {"opt_in": {}, "daily_cap": 0}, 3, 99)[0] == ["sms"] # OTP at 3 a.m. still goes
142
143 # 4. OTPs
144 import random
145 rnd = random.Random(33)
146 svc = OtpService()
147 code, _ = svc.request("+91-98", now=0, rnd=rnd)
148 assert svc.verify("+91-98", "000000" if code != "000000" else "111111", now=10) == (False, "wrong code")
149 assert svc.verify("+91-98", code, now=20) == (True, "ok")
150 assert svc.verify("+91-98", code, now=21) == (False, "no OTP") # single use
151 code, _ = svc.request("+91-98", now=30, rnd=rnd)
152 assert svc.verify("+91-98", code, now=400) == (False, "expired")
153 assert svc.request("+91-98", now=500, rnd=rnd)[1] == "sent"
154 assert svc.request("+91-98", now=600, rnd=rnd)[1] == "too many OTPs: try later" # 3 in an hour
155 svc2 = OtpService()
156 code, _ = svc2.request("+44-77", now=0, rnd=rnd)
157 for _ in range(5):
158 svc2.verify("+44-77", "999999" if code != "999999" else "888888", now=1)
159 assert svc2.verify("+44-77", code, now=2) == (False, "locked: too many attempts")
160 assert "482913" not in str(svc.store) # only hashes stored
161 assert not should_still_send(created_at=0, now=400, ttl=300)
162
163 # guessing odds with 5 attempts on a 6-digit code
164 assert 5 / 1_000_000 == 0.000005
165
166 return {"shared_wait_s": 400, "own_wait_s": 0.1, "delivery": how, "dup": how2}
167
168
169if __name__ == "__main__":
170 print("All checks pass.", run_checks())