Chat Bot System Refactoring

The Challenge You have an old codebase for a chat service. It supports different bots (AwayBot, MeetBot, TacoBot) that react to specific commands. Right now, all the code is mixed together in one b

The Challenge You have an old codebase for a chat service. It supports different bots (AwayBot, MeetBot, TacoBot) that react to specific commands. Right now, all the code is mixed together in one big function. It uses global variables, which is messy. Your job is to refactor (clean up) this code. You need to make it easy to read, manage, and add new bots to later. This problem tests if you can: - Find bad coding habits ("code smells"). - Use Object-Oriented Design. - Create simple interfaces. - Manage how data moves between parts of the code. - Write code that is easy to test. ## Part 1: Looking at the Old Code ### The Legacy Code Here is the code you need to fix:

python aways: dict[str, str] = {} tacos: dict[str, int] = {} messages = [] def sendMessage(name: str, msg: str) -> None: messages.append(name + ": " + msg) # AwayBot logic for away, away_msg in aways.items(): if away in msg: messages.append(f"AwayBot: {away} is away: {away_msg}") # MeetBot logic if msg[1:5] == "meet": messages.append( "MeetBot: Google Meet with @" + name + ", and " + msg[6:] + " starting at <a href="https://meet.google.com/abc-def-123" target="_blank" rel="noopener" class="text-emerald-400 hover:text-emerald-300 underline">https://meet.google.com/abc-def-123</a>" ) aways[name] = "@" + name + " may be in a meeting right now" aways[msg[7:]] = "@" + msg[7:] + " may be in a meeting right now" # TacoBot logic if msg[1:9] == "givetaco": num_tacos = len(msg.split(" ")[1]) who = msg.split(" ")[2] if who[1:] not in tacos: tacos[who[1:]] = 0 tacos[who[1:]] += num_tacos messages.append( "TacoBot: @" + name + " gave @" + who + " " + str(num_tacos) + " 🌮's. " + who + f" now have {tacos[who[1:]]} 🌮s." ) # Away status logic if msg[1:5] == "away": aways[name] = msg[6:]

Required Output Your new code must produce the exact same output for these inputs:

python sendMessage(name="Alice", msg="Hello") sendMessage(name="Bob", msg="Hi") sendMessage(name="Alice", msg="Nice job on your presentations") sendMessage(name="Cindy", msg="/givetaco 🌮🌮 @justin") sendMessage(name="Alice", msg="Bob let's meet") sendMessage(name="Bob", msg="/meet Alice") sendMessage(name="David", msg="/away out for lunch") sendMessage(name="Emily", msg="Anyone around?") sendMessage(name="Frank", msg="/meet David") assert messages == ["Alice: Hello", "Bob: Hi", "Alice: Nice job on your presentations", "Cindy: /givetaco 🌮🌮 @justin", "TacoBot: @Cindy gave @@justin 2 🌮's. @justin now have 2 🌮s.", "Alice: Bob let's meet", "Bob: /meet Alice", "MeetBot: Google Meet with @Bob, and Alice starting at <a href="https://meet.google.com/abc-def-123" target="_blank" rel="noopener" class="text-emerald-400 hover:text-emerald-300 underline">https://meet.google.com/abc-def-123</a>", "David: /away out for lunch", "Emily: Anyone around?", "Frank: /meet David", "AwayBot: David is away: out for lunch", "MeetBot: Google Meet with @Frank, and David starting at <a href="https://meet.google.com/abc-def-123" target="_blank" rel="noopener" class="text-emerald-400 hover:text-emerald-300 underline">https://meet.google.com/abc-def-123</a>"]

What the Bots Do 1.

AwayBot: - Watches for mentions of users who are away. - Replies with that user's away message. - Users set their status with /away <message>. 2.

MeetBot: - Starts when it sees /meet <username>. - Makes a meeting link. - Sets both users' status to "may be in a meeting right now". 3.

TacoBot: - Starts when it sees /givetaco <tacos> <@username>. - Counts the taco emojis. - Keeps a total count of tacos for each user. - Announces the gift and the new total. ### Why the Old Code is Bad Before fixing it, let's list the problems: -

Tight coupling: All logic is stuck in one function. -

Global state: aways and tacos are global variables. -

Hard to extend: Adding a new bot means changing the main function. -

Messy logic: Parsing strings and business rules are mixed. -

Hard to test: You cannot test just one bot by itself. -

No validation: It assumes every command is typed perfectly. -

Bad readability: String slicing makes it hard to read. -

Dependencies: It is hard for bots to talk to each other cleanly. ## Part 2: First Solution - Using Interfaces ### Refactoring Goals Create an interface named IBotService. It needs two methods: 1. shouldActivate: Check if the bot needs to run. 2. execute: Run the bot logic and return a list of messages. You need to: - Make a separate class for each bot. - Register these bots in a chat room class. - Loop through the bots when a message comes in.

Solution: Using Interfaces

python from abc import ABC, abstractmethod from typing import List class IBotService(ABC): """Interface for chat bots""" @abstractmethod def should_activate(self, name: str, msg: str) -> bool: """Check if this bot should respond to the message""" pass @abstractmethod def execute(self, name: str, msg: str) -> List[str]: """Execute bot logic and return response messages""" pass class ChatRoom: def __init__(self): self.bots: List[IBotService] = [] self.messages: List[str] = [] def register_bot(self, bot: IBotService): """Register a bot with the chat room""" self.bots.append(bot) def send_message(self, name: str, msg: str): """Process a message through all registered bots""" # Add the user's message self.messages.append(f"{name}: {msg}") # Check each bot for bot in self.bots: if bot.should_activate(name, msg): bot_responses = bot.execute(name, msg) self.messages.extend(bot_responses) class MeetBot(IBotService): def __init__(self, away_bot=None): self.away_bot = away_bot # Pass AwayBot so we can update status def should_activate(self, name: str, msg: str) -> bool:

**return** msg.startswith("/meet ") def execute(self, name: str, msg: str) -> List[str]: # Parse the command parts = msg.split(" ", 1) if len(parts) < 2:

**return** [] other_user = parts[1] # Generate meeting link response = f"MeetBot: Google Meet with @{name}, and {other_user} starting at <a href="https://meet.google.com/abc-def-123" target="_blank" rel="noopener" class="text-emerald-400 hover:text-emerald-300 underline">https://meet.google.com/abc-def-123</a>" # Update away status through AwayBot if self.away_bot: self.away_bot.set_away(name, f"@{name} may be in a meeting right now") self.away_bot.set_away(other_user, f"@{other_user} may be in a meeting right now")

**return** [response] class TacoBot(IBotService): def __init__(self): self.taco_counts: dict[str, int] = {} def should_activate(self, name: str, msg: str) -> bool:

**return** msg.startswith("/givetaco ") def execute(self, name: str, msg: str) -> List[str]: # Parse: /givetaco 🌮🌮 @username parts = msg.split(" ") if len(parts) < 3:

**return** [] taco_string = parts[1] recipient = parts[2] # Count tacos and fix username format num_tacos = len(taco_string) username = recipient.lstrip("@") # Update count if username not in self.taco_counts: self.taco_counts[username] = 0 self.taco_counts[username] += num_tacos # Format response response = f"TacoBot: @{name} gave @{recipient} {num_tacos} 🌮's. {recipient} now have {self.taco_counts[username]} 🌮s."

**return** [response] class AwayBot(IBotService): def __init__(self): self.away_statuses: dict[str, str] = {} def should_activate(self, name: str, msg: str) -> bool: # Check if setting away status if msg.startswith("/away "):

**return** True # Check if message mentions someone who is away for away_user in self.away_statuses: if away_user in msg:

**return** True return False def execute(self, name: str, msg: str) -> List[str]: responses = [] # Handle /away command if msg.startswith("/away "): away_message = msg[6:] # Everything after "/away " self.set_away(name, away_message)

**return** [] # /away command doesn't produce a message # Check for mentions of away users for away_user, away_msg in self.away_statuses.items(): if away_user in msg: responses.append(f"AwayBot: {away_user} is away: {away_msg}")

**return** responses def set_away(self, username: str, message: str): """Public method for other bots to set away status""" self.away_statuses[username] = message # Usage chat_room = ChatRoom() # Create bots with dependencies away_bot = AwayBot() meet_bot = MeetBot(away_bot=away_bot) taco_bot = TacoBot() # Register bots chat_room.register_bot(away_bot) chat_room.register_bot(meet_bot) chat_room.register_bot(taco_bot) # Send messages chat_room.send_message("Alice", "Hello") chat_room.send_message("Cindy", "/givetaco 🌮🌮 @justin") # ... etc

Why This Design Works 1.

Dependency Injection: MeetBot is given access to AwayBot so it can change status. 2.

Encapsulation: Each bot keeps its own data private (like taco counts). 3.

Single Responsibility: Each class does only one thing. 4.

Open/Closed Principle: You can add new bots without touching the old ones. 5.

Testability: You can test one bot without running the others. ## Part 3: Advanced Solution - Using Events The interviewer might say that passing bots into other bots (Dependency Injection) gets messy if you have many bots. They might ask for an

Event-Driven Architecture. ### New Requirements Refactor the code so: - Bots publish "Events". - Other bots listen (subscribe) to these events. - Bots do not need to know about each other directly. - A central "Event Bus" handles the communication. ### Code: Event-Driven Approach

python from abc import ABC, abstractmethod from typing import List, Callable, Dict from dataclasses import dataclass @dataclass class Event: """Base class for events""" event_type: str data: dict class EventBus: """Central hub for bot communication""" def __init__(self): self.subscribers: Dict[str, List[Callable]] = {} def subscribe(self, event_type: str, handler: Callable[[Event], None]): """Listen for a specific event type""" if event_type not in self.subscribers: self.subscribers[event_type] = [] self.subscribers[event_type].append(handler) def publish(self, event: Event): """Send an event to all listeners""" if event.event_type in self.subscribers: for handler in self.subscribers[event.event_type]: handler(event) class IBotService(ABC): """Interface for chat bots""" def __init__(self, event_bus: EventBus): self.event_bus = event_bus self.setup_subscriptions() def setup_subscriptions(self): """Override to subscribe to events""" pass @abstractmethod def should_activate(self, name: str, msg: str) -> bool: pass @abstractmethod def execute(self, name: str, msg: str) -> List[str]: pass class AwayBot(IBotService): def __init__(self, event_bus: EventBus): self.away_statuses: dict[str, str] = {} super().__init__(event_bus) def setup_subscriptions(self): # Listen for meeting events to set away status self.event_bus.subscribe("user_meeting_started", self.handle_meeting_started) def handle_meeting_started(self, event: Event): """Handle when a user starts a meeting""" username = event.data["username"] self.away_statuses[username] = f"@{username} may be in a meeting right now" def should_activate(self, name: str, msg: str) -> bool: if msg.startswith("/away "):

**return** True for away_user in self.away_statuses: if away_user in msg:

**return** True return False def execute(self, name: str, msg: str) -> List[str]: responses = [] if msg.startswith("/away "): away_message = msg[6:] self.away_statuses[name] = away_message # Publish event self.event_bus.publish(Event( event_type="user_away_status_changed", data={"username": name, "message": away_message} ))

**return** [] for away_user, away_msg in self.away_statuses.items(): if away_user in msg: responses.append(f"AwayBot: {away_user} is away: {away_msg}")

**return** responses class MeetBot(IBotService): def setup_subscriptions(self): # MeetBot doesn't need to listen to events here pass def should_activate(self, name: str, msg: str) -> bool:

**return** msg.startswith("/meet ") def execute(self, name: str, msg: str) -> List[str]: parts = msg.split(" ", 1) if len(parts) < 2:

**return** [] other_user = parts[1] # Publish events for both participants self.event_bus.publish(Event( event_type="user_meeting_started", data={"username": name} )) self.event_bus.publish(Event( event_type="user_meeting_started", data={"username": other_user} )) response = f"MeetBot: Google Meet with @{name}, and {other_user} starting at <a href="https://meet.google.com/abc-def-123" target="_blank" rel="noopener" class="text-emerald-400 hover:text-emerald-300 underline">https://meet.google.com/abc-def-123</a>"

**return** [response] class TacoBot(IBotService): def __init__(self, event_bus: EventBus): self.taco_counts: dict[str, int] = {} super().__init__(event_bus) def setup_subscriptions(self): pass # TacoBot doesn't listen to events def should_activate(self, name: str, msg: str) -> bool:

**return** msg.startswith("/givetaco ") def execute(self, name: str, msg: str) -> List[str]: parts = msg.split(" ") if len(parts) < 3:

**return** [] taco_string = parts[1] recipient = parts[2] num_tacos = len(taco_string) username = recipient.lstrip("@") if username not in self.taco_counts: self.taco_counts[username] = 0 self.taco_counts[username] += num_tacos # Publish event self.event_bus.publish(Event( event_type="tacos_given", data={ "giver": name, "recipient": username, "count": num_tacos, "total": self.taco_counts[username] } )) response = f"TacoBot: @{name} gave @{recipient} {num_tacos} 🌮's. {recipient} now have {self.taco_counts[username]} 🌮s."

**return** [response] class ChatRoom: def __init__(self): self.event_bus = EventBus() self.bots: List[IBotService] = [] self.messages: List[str] = [] def register_bot(self, bot: IBotService): self.bots.append(bot) def send_message(self, name: str, msg: str): self.messages.append(f"{name}: {msg}") for bot in self.bots: if bot.should_activate(name, msg): bot_responses = bot.execute(name, msg) self.messages.extend(bot_responses) # Usage chat_room = ChatRoom() # Create bots with shared event bus away_bot = AwayBot(chat_room.event_bus) meet_bot = MeetBot(chat_room.event_bus) taco_bot = TacoBot(chat_room.event_bus) # Register bots chat_room.register_bot(away_bot) chat_room.register_bot(meet_bot) chat_room.register_bot(taco_bot)

Why Use Events? 1.

Loose Coupling: Bots do not need to link to each other. 2.

Scalability: You can add many new bots that listen to the same events. 3.

Flexibility: An event can trigger actions in multiple different bots. 4.

Testability: You can fake the events to test how a bot reacts. ## Part 4: Testing the Code The interviewer may ask you to prove your code works by writing tests. ### Testing Individual Bots

python import unittest class TestTacoBot(unittest.TestCase): def setUp(self): self.event_bus = EventBus() self.taco_bot = TacoBot(self.event_bus) def test_should_activate_on_givetaco_command(self): self.assertTrue(self.taco_bot.should_activate("Alice", "/givetaco 🌮🌮 @bob")) self.assertFalse(self.taco_bot.should_activate("Alice", "Hello")) def test_execute_gives_tacos(self): responses = self.taco_bot.execute("Alice", "/givetaco 🌮🌮🌮 @bob") self.assertEqual(len(responses), 1) self.assertIn("3 🌮's", responses[0]) self.assertIn("now have 3 🌮s", responses[0]) def test_accumulates_tacos(self): self.taco_bot.execute("Alice", "/givetaco 🌮🌮 @bob") responses = self.taco_bot.execute("Charlie", "/givetaco 🌮 @bob") self.assertIn("now have 3 🌮s", responses[0]) class TestMeetBot(unittest.TestCase): def setUp(self): self.event_bus = EventBus() self.meet_bot = MeetBot(self.event_bus) # Keep track of published events self.published_events = [] self.event_bus.subscribe("user_meeting_started", lambda e: self.published_events.append(e)) def test_creates_meeting_link(self): responses = self.meet_bot.execute("Alice", "/meet Bob") self.assertEqual(len(responses), 1) self.assertIn("Google Meet with @Alice, and Bob", responses[0]) self.assertIn("<a href="https://meet.google.com" target="_blank" rel="noopener" class="text-emerald-400 hover:text-emerald-300 underline">https://meet.google.com</a>", responses[0]) def test_publishes_meeting_events(self): self.meet_bot.execute("Alice", "/meet Bob") self.assertEqual(len(self.published_events), 2) self.assertEqual(self.published_events[0].data["username"], "Alice") self.assertEqual(self.published_events[1].data["username"], "Bob") class TestAwayBot(unittest.TestCase): def setUp(self): self.event_bus = EventBus() self.away_bot = AwayBot(self.event_bus) def test_set_away_status(self): responses = self.away_bot.execute("David", "/away out for lunch") # /away doesn't produce an output message self.assertEqual(len(responses), 0) self.assertIn("David", self.away_bot.away_statuses) def test_notify_when_away_user_mentioned(self): self.away_bot.execute("David", "/away out for lunch") responses = self.away_bot.execute("Alice", "Hey David, are you around?") self.assertEqual(len(responses), 1) self.assertIn("David is away", responses[0]) def test_handles_meeting_started_event(self): event = Event( event_type="user_meeting_started", data={"username": "Alice"} ) self.away_bot.handle_meeting_started(event) self.assertIn("Alice", self.away_bot.away_statuses) self.assertIn("may be in a meeting", self.away_bot.away_statuses["Alice"])

Testing the Whole System

python class TestChatRoomIntegration(unittest.TestCase): def setUp(self): self.chat_room = ChatRoom() away_bot = AwayBot(self.chat_room.event_bus) meet_bot = MeetBot(self.chat_room.event_bus) taco_bot = TacoBot(self.chat_room.event_bus) self.chat_room.register_bot(away_bot) self.chat_room.register_bot(meet_bot) self.chat_room.register_bot(taco_bot) def test_full_conversation_flow(self): self.chat_room.send_message("Alice", "Hello") self.chat_room.send_message("Bob", "/meet Alice") self.chat_room.send_message("Charlie", "Hey Bob") messages = self.chat_room.messages # Check user messages self.assertIn("Alice: Hello", messages) self.assertIn("Bob: /meet Alice", messages) # Check MeetBot response self.assertTrue(any("MeetBot" in msg and "Google Meet" in msg for msg in messages)) # Check AwayBot response (Bob is in a meeting) self.assertTrue(any("AwayBot" in msg and "Bob is away" in msg for msg in messages))

Part 5: Bonus Questions ### Extension 1: Add Validations Make sure commands are typed correctly before running them:

python class Command: """Helper for parsed commands""" def __init__(self, raw_msg: str): self.raw_msg = raw_msg self.is_command = raw_msg.startswith("/") if self.is_command: parts = raw_msg[1:].split(" ", 1) self.command_name = parts[0] self.args = parts[1] if len(parts) > 1 else "" else: self.command_name = None self.args = None def validate_format(self, expected_arg_count: int) -> bool: """Check if command has enough arguments""" if not self.is_command:

**return** False arg_parts = self.args.split() if self.args else []

**return** len(arg_parts) >= expected_arg_count class TacoBot(IBotService): def execute(self, name: str, msg: str) -> List[str]: cmd = Command(msg) if not cmd.validate_format(2):

**return** ["TacoBot: Invalid command. Usage: /givetaco <tacos> <@username>"] # ... rest of logic

Extension 2: Control Bot Order Sometimes one bot needs to run before another (e.g., check for meetings before checking away status).

python class IBotService(ABC): @property @abstractmethod def priority(self) -> int: """Lower numbers run first""" pass class ChatRoom: def send_message(self, name: str, msg: str): self.messages.append(f"{name}: {msg}") # Sort bots by priority sorted_bots = sorted(self.bots, key=lambda b: b.priority) for bot in sorted_bots: if bot.should_activate(name, msg): bot_responses = bot.execute(name, msg) self.messages.extend(bot_responses)

Extension 3: Async Operations If you need to talk to a slow database or API, use async code:

python import asyncio class AsyncEventBus: async def publish(self, event: Event): """Publish event asynchronously""" if event.event_type in self.subscribers: await asyncio.gather(*[handler(event) for handler in self.subscribers[event.event_type]]) class IBotService(ABC): @abstractmethod async def execute(self, name: str, msg: str) -> List[str]: """Execute bot logic asynchronously""" pass

Extension 4: Middleware Pipeline Add steps like logging or rate limiting (spam prevention) before the bots see the message:

python class Middleware(ABC): @abstractmethod def process_message(self, name: str, msg: str, next_handler): pass class LoggingMiddleware(Middleware): def process_message(self, name: str, msg: str, next_handler): print(f"[LOG] {name}: {msg}")

**return** next_handler(name, msg) class RateLimitMiddleware(Middleware): def __init__(self): self.message_counts = {} def process_message(self, name: str, msg: str, next_handler): # Check spam rules if self.is_rate_limited(name):

**return** ["Rate limit exceeded. Please wait."]

**return** next_handler(name, msg)

Review and Wrap-up ### Common Errors to Avoid 1.

Breaking Tests: Always make sure the original test cases still pass. 2.

Wrong Order: Messages must appear in the correct sequence. 3.

Double Alerts: AwayBot should not reply twice to the same message. 4.

Parsing Bugs: Be careful if a command like /meet has no username after it. 5.

Leaking State: One test should not affect the results of the next test. ### Tricky Scenarios Think about these edge cases: 1.

Empty commands: Someone types /meet with no name. 2.

Self-mentions: A user mentions themselves while they are away. 3.

Multiple mentions: A message mentions two different people who are away. 4.

Weird characters: Using emojis or unicode in names. 5.

Capitalization: Should /MEET work the same as /meet? ### Topics for Discussion Be ready to talk about these trade-offs: 1.

Event Bus vs. Dependency Injection: - Events: Flexible, but can be hard to track what is happening. - Dependency Injection: Simple to understand, but gets messy with too many connections. 2.

Global State vs. Classes: - Global State: Easy to write quickly, but very bad for maintenance. - Classes: Keep data safe inside the object (Encapsulation). 3.

Sync vs. Async: - Sync: Easier to test and write. - Async: Necessary for real production apps that use databases or internet calls.