System Requirements

Functional -

1. Messaging should have the features listed below.
   1. Send
   2. Receive
   3. Delete
   4. Search
   5. React
2. We need all chats to be listed in the app
3. We indicate if a user is currently online or not
4. We need to indicate to a user if the user to which they've sent a message has read it or not
5. We need to implement a notification system
6. User authentication is not covered in this design
7. The ability to add/remove friends is also out of the scope of this design

Non-Functional -

1. Real Time - Users need to be able to char in real time
2. Scalability - Ability to be able to handle 1 billion users at once
3. Availability - We aim for virtually no downtime
4. Reliability - Messages can be delayed but never lost
5. Consistency - We are fine with trading off a bit of consistency for availability. We aim for eventual consistency
6. Fault Tolerance - Messages should continue even in case some parts of the network is down for some reason

Capacity Estimation

1. We assume 1 billion (10^9) daily active users
2. We assume that each user sends 10 messages on a day, totaling 10^10 messages in a day
3. In a single second, we do 10 ^ 10 / 10^5 transactions = 10^5 transactions per second
4. Peak transactions per second = 5 * 10^5
5. We assume that each message takes 1 kilobyte of data, thus to store 10^10 messages, we need 10^10 * 10^3 = 10^13 bytes of data = 10 TB
6. In a month we would need 300 TB data, in one year we would need 3600 TB data, in 10 years we would need 36000 TB data = 36 PB data
7. Factoring in redundancy, we would need at least 100 PB data over 10 years
8. We also assume that the metadata for each user takes 1 KB data, for 1 billion users, we would need 10^12 bytes of data = 1 TB data

API Design

1. Send Message
   1. Parameters of the request are given below.
      1. unique_id
      2. sender_id
      3. receiver_id
      4. message_content
      5. timestamp
2. Get Messages
   1. Parameters of the request are given below.
      1. unique_id
      2. user_id
      3. timestamp
   2. All timestamps that are newer than the given timestamp in the request are fetched
3. Get Status
   1. Parameters of the request are given below
      1. unique_id
      2. friend_ids
   2. User sends a list of userIDs whose status they want to see. The user gets a JSON file with all the users and their online status
4. Search Chat
   1. Parameters of the request are given below.
      1. unique_id
      2. user_id
      3. search_terms
      4. timestamp
   2. Returns the message IDs that are relevant to the search term

Database Design

1. There will be two databases here, one for the user metadata and the other for actually storing the messages. For the metadata, we can use a relational database and for the messages, we use a non-relational database like a key-value store
2. The relational database would have the user table whose structure is given below.
   1. user
      1. id Primary Key
      2. username
      3. email
      4. password (Encrypted)
      5. created_at
3. The non-relational database would have a table with the given columns. For the database choice, we use Cassandra
   1. messages
      1. message_id
      2. from_user Foreign Key
      3. to_user Foreign Key
      4. content
      5. created_at Secondary Partition Key
      6. is_read
   2. We put indexes on both the from_user and to_user fields since those fields will be used the most when doing reads and writes
   3. created_at will help us get messages in a range
4. For the presence part, we store the information in an in-memory database like Redis, whose structure could be like given below.
   1. presence
      1. user_id
      2. status
      3. last_seen
   2. Similarly, we use a fast in-memory database to keep track of the server a user is connected to, to initiate web socket connections to that server in case of a disconnection. Again, something like Redis can be used here
      1. user_server
         1. user_id
         2. server_id

High-level Design

flowchart LR subgraph message_search client1[user1] --> |search 'monday'| LB{API GateWay} LB --> SearchService[Search Service] SearchService --> SC[(SearchCache)] SC --> SearchService SearchService --> EC1[(Elasticsearch)] end subgraph message_storage TT[ChatService] --> |enqueue msg| Q[Kafka - partitioned and replicated] subgraph workers W[worker1] WW[worker2] WWW[worker3] end subgraph messageCluster M((Messages Node1)) MM((Messages Node2)) M <--> MM MM <--> M end W --> Q WW --> Q W --> M WW --> MM SS[Sparq Job] --> Q SS --> EC[(Elasticsearch)] end subgraph message_send subgraph clients A[web] B[mobile] end subgraph websocket servers G[Server2] GG[Server1] end A --> |send msg to userB| GG GG --> E[ChatService] E --> EE[(UserServerDB)] EE --> |userB in server2| E E --> G[Server2] E --> |save| D[(Messages)] G --> |msg from userA over websocket| B end

Request Flows

The high level design has all the flows demonstrated.

Detailed Component Design

1. Sending Messages
   1. To exchange messages between users, we use web socket connections
   2. When a user comes online, they will be paired via a web socket connection to a server
   3. This user server pairing will be stored in the user_server table
   4. When a message is needed to be sent to a user, the server to which the user is connected looks up which server the recipient user is connected and then forwards the message to that server, which is then routed to the recipient
   5. Also, the message is added to a message queue, which then proceeds to being stored in the key-value message store, which is Cassandra in our case
2. Message Storage
   1. Apart from forwarding messages to the other user, we also want to be storing messages in a durable storage to protect them from being lost during failures
   2. To write messages to our non-relational database, we use a log based messaging queue like Apache Kafka
   3. Using Apache Kakfa ensures that messages are received in order and are stored in the messaging queue in case the database is not available for writes
   4. We will have a bunch of workers that will save the messages to the database
   5. Cassandra is utilized here since it is highly scalable and is able to handle a large volume of writes well
3. Message Search
   1. For searches, we utilize Elastic Search. We set a cache where we cache the recent search results, we hit the cache first, in case the result is found there, we return the search result, otherwise, we hot our Elastic Search server
   2. Elastic Search is based on Apache's Lucene and uses a reverse index to do text based searches. We map a word to the to message IDs in which the word occurs. In case the user's search term involves multiple words, we fetch the result based on each search word and merge results

Trade offs/Tech choices

1. Database
   1. For the relational database, we can use something like PostgreSQL since we are not going to be running into scenarios where there are a bunch of write conflicts
   2. For the message store, we use Cassandra since it is highly scalable and will be able to handle the large volume of writes. Also, it uses consistent hashing which means our message data will be spread evenly across all the data nodes. It uses an Log Structures Merge (LSM) tree coupled with a write ahead log (WAL) to do writes
2. Message Queue
   1. We have chosen Kakfa since we want something log based to preserve the order of messages and also, Kakfa stores messages on disk meaning we do not lose them in case our database is not available for writes or if our messaging queue server goes down
3. Connection
   1. Web Socket is used here instead of polling or long polling since those two do not support bi-directional communication and in case of long polling, there is no way to tell if the connection is active or not
   2. Maintaining connections via web-socket is easier than creating new connections using long polling

Failure Scenarios / Bottlenecks

1. We need to be able to reconnect a user who has lost connection gracefully in case of a reconnection
2. Message database cluster should be built robustly since we cannot afford to lost messages in any scenario. We also need to replicate the nodes in a different data-center so that we do not end up losing messages
3. Number of servers dedicated for web server connections
   1. Since there are millions of users connected at once, one server can only handle around 50 thousand web-socket connections at a time we need 20 servers to handle web socket connections for 1 million concurrent users, for 200 million active users we would need 4000 servers
   2. Fan-out from the chat service to these servers is a bottleneck as well, since the chat service needs to query where to push the message, the Kafka queue needs to be able to handle this throughput
   3. The number of workers that pull from the queue and then send them to the proper server is a number we need to track as well which is also a bottleneck

Future Improvements

1. Multi region and multi data-center deployments. This will reroute users from different countries to their own data-centers. This will make it more complicated to chat with users from other countries but it is doable since we store where the users are connecting to using their web socket
2. Using Redis to manage sessions to that we can figure out if users are active or not
3. Make the sending message feature more robust, for example is the user is offline/online
4. Make re-connections more robust, for example if we already have a session (when we check Redis) we reconnect to the same server so we do not have to do a full handshake again
5. Have proper dashboard, notifications, logging for the whole system
   1. We will have to capture metrics from all the components so we can judge the health of the system and setup alerts when things are unhealthy. Examples of things to monitor are given here.
      1. Message queues are getting filled faster than the workers can work on de-queuing the messages
      2. CPU usage on every worker and servers so we can see how busy everyone is
      3. Cache misses so we can improve our caching strategies
      4. Latency on end to end transactions / API calls
      5. Error codes and sub-codes
   2. To track metrics, we can use a metrics monitoring and logging system like Grafana or Prometheus
   3. API Tracing
      1. Build tools to trace network requests so we can visualize the flow and see where the error came from
   4. Distributed Logs
      1. Setup a system to collect logs from different servers and funnel them to a system that parses through them
      2. Developers can search through them when they are debugging issues
      3. We can use something like Datadog or Azure logs