1"""Chapter 38 (case study: chat app like WhatsApp) deep-dive code.
21. Gateway routing: which gateway holds which user's connection.
32. Message IDs and ordering: client IDs for de-duplication, server sequence numbers per chat.
43. Ticks: sent -> delivered -> read, per recipient device.
54. Offline delivery: a per-device inbox read from a cursor.
65. Group chat: fan-out cost, and receipts that grow with group size.
7Run: python3 ch38_chat.py"""
8from collections import defaultdict
9
10
11# ---- 1. routing ---------------------------------------------------------------------
12
13class Registry:
14 def __init__(self):
15 self.where = {} # (user, device) -> gateway
16
17 def connect(self, user, device, gw):
18 self.where[(user, device)] = gw
19
20 def disconnect(self, user, device):
21 self.where.pop((user, device), None)
22
23 def online_devices(self, user):
24 return {d: g for (u, d), g in self.where.items() if u == user}
25
26
27# ---- 2-4. the chat server ----------------------------------------------------------
28
29class ChatServer:
30 def __init__(self, registry):
31 self.reg = registry
32 self.seq = defaultdict(int) # chat -> last sequence number
33 self.log = defaultdict(list) # chat -> [(seq, client_id, sender, body)]
34 self.by_client_id = {} # (chat, sender, client_id) -> seq
35 self.inbox_cursor = defaultdict(int) # (user, device, chat) -> last seq delivered
36 self.status = {} # (chat, seq, user, device) -> sent|delivered|read
37 self.members = {}
38 self.pushed = [] # (user, device, chat, seq) sent live
39
40 def create_chat(self, chat, members):
41 self.members[chat] = list(members)
42
43 def send(self, chat, sender, client_id, body):
44 key = (chat, sender, client_id)
45 if key in self.by_client_id: # client retried after a lost ack
46 return self.by_client_id[key], "duplicate"
47 self.seq[chat] += 1
48 s = self.seq[chat]
49 self.log[chat].append((s, client_id, sender, body))
50 self.by_client_id[key] = s
51 for m in self.members[chat]:
52 if m == sender:
53 continue
54 for device in ["phone", "laptop"]:
55 self.status[(chat, s, m, device)] = "sent"
56 for device in self.reg.online_devices(m):
57 self.pushed.append((m, device, chat, s))
58 return s, "stored" # the sender sees one tick
59
60 def ack_delivered(self, user, device, chat, upto_seq):
61 self.inbox_cursor[(user, device, chat)] = max(self.inbox_cursor[(user, device, chat)], upto_seq)
62 for (c, s, u, d), st in list(self.status.items()):
63 if c == chat and u == user and d == device and s <= upto_seq and st == "sent":
64 self.status[(c, s, u, d)] = "delivered"
65
66 def mark_read(self, user, chat, upto_seq):
67 for (c, s, u, d), st in list(self.status.items()):
68 if c == chat and u == user and s <= upto_seq:
69 self.status[(c, s, u, d)] = "read"
70
71 def sync(self, user, device, chat):
72 """A device that was offline asks: everything after my cursor, in order."""
73 after = self.inbox_cursor[(user, device, chat)]
74 return [(s, body) for s, _, _, body in self.log[chat] if s > after]
75
76 def ticks(self, chat, seq):
77 """What the sender sees: one tick (stored), two grey (all devices of all recipients got it), blue (all read)."""
78 sts = [st for (c, s, u, d), st in self.status.items() if c == chat and s == seq]
79 if sts and all(st == "read" for st in sts):
80 return "blue"
81 if sts and all(st in ("delivered", "read") for st in sts):
82 return "two grey"
83 return "one grey"
84
85
86# ---- 5. groups --------------------------------------------------------------------
87
88def group_costs(members, messages_per_day):
89 deliveries = (members - 1) * messages_per_day
90 receipts = 2 * (members - 1) * messages_per_day # delivered + read, per recipient
91 return deliveries, receipts
92
93
94# ---- checks ------------------------------------------------------------------------
95
96def run_checks():
97 reg = Registry()
98 reg.connect("asha", "phone", "gw-3")
99 reg.connect("ravi", "phone", "gw-7")
100 reg.connect("ravi", "laptop", "gw-2")
101 assert reg.online_devices("ravi") == {"phone": "gw-7", "laptop": "gw-2"}
102
103 srv = ChatServer(reg)
104 srv.create_chat("c1", ["asha", "ravi"])
105
106 # 2. ordering and de-duplication
107 s1, how1 = srv.send("c1", "asha", "a-001", "Leaving now")
108 s2, how2 = srv.send("c1", "asha", "a-002", "Reach in 10 min")
109 s2b, how2b = srv.send("c1", "asha", "a-002", "Reach in 10 min") # retry after a lost ack
110 assert (s1, s2) == (1, 2) and how2b == "duplicate" and s2b == 2
111 assert len(srv.log["c1"]) == 2
112 assert [m for m in srv.pushed if m[0] == "ravi"] == [("ravi", "phone", "c1", 1), ("ravi", "laptop", "c1", 1),
113 ("ravi", "phone", "c1", 2), ("ravi", "laptop", "c1", 2)]
114
115 # 3. ticks
116 assert srv.ticks("c1", 1) == "one grey"
117 srv.ack_delivered("ravi", "phone", "c1", 2)
118 assert srv.ticks("c1", 1) == "one grey" # laptop hasn't got it yet
119 srv.ack_delivered("ravi", "laptop", "c1", 2)
120 assert srv.ticks("c1", 1) == "two grey"
121 srv.mark_read("ravi", "c1", 1)
122 assert srv.ticks("c1", 1) == "blue" and srv.ticks("c1", 2) == "two grey"
123
124 # 4. offline delivery: Ravi's laptop goes offline, misses 3 messages, syncs in order
125 reg.disconnect("ravi", "laptop")
126 for i in range(3):
127 srv.send("c1", "asha", f"a-10{i}", f"msg {i}")
128 assert [s for s, _ in srv.sync("ravi", "laptop", "c1")] == [3, 4, 5]
129 srv.ack_delivered("ravi", "laptop", "c1", 5)
130 assert srv.sync("ravi", "laptop", "c1") == []
131
132 # 5. groups: a 1,024-member group sending 200 messages a day
133 deliveries, receipts = group_costs(1_024, 200)
134 assert deliveries == 204_600 and receipts == 409_200
135 # Step 2 arithmetic: 500M daily users x 40 messages
136 msgs = 500_000_000 * 40
137 assert msgs == 20e9 and round(msgs / 86_400) == 231_481
138 assert round(msgs * 100 / 1e12) == 2 # TB a day of message text at ~100 B
139 assert 200_000_000 / 100_000 == 2_000 # gateways for 200M connections at 100k each (Ch 21)
140
141 return {"seq": (s1, s2), "dup": how2b, "offline_sync": [3, 4, 5], "group": (deliveries, receipts)}
142
143
144if __name__ == "__main__":
145 print("All checks pass.", run_checks())