Loading...
DAU - 0.5 billion
messages per user - 50
average text payload - 1KB
Storage day - 0.5*10^9 * 50 * 1KB = 25TB -> 25*365 = 9PB year
Bandwidth - 25GB/(24*60*60) * 8 = 2.3Gbps
QPS - 0.5*10^9 * 50 / (24*60*60) = 290k/sec
Server - 0.5*10^9/64k = 7812 servers for peak load
POST /api/v1/{conversation_id}/message
{
auth_token,
text
}
POST /api/v1/messages/{msg_id}/ack
{
auth_token,
message_id
}
GET /api/v1/{conversation_id}/messages?offest&limit - returns
[message_ids]
GET /api/v1/messages/{msg_id} - returns
{
text
}
POST /api/v1/presence/heartbeat
{
auth_token
}
GET /api/v1/presence/{user_id}
users table SQL
{
user_id,
name,
email,
}
groups table SQL
{
pk group_id
created
}
group_membership table SQL
{
pk group_id, user_id
}
The users table will be a SQL since it needs to have acid properties like unique user_id and email and foreign key for user_id and group_id to refer to a valid user and group. The users, groups, and group membership tables will be sharded according to user_id.
I will add an index on user_id column in the group_membership table, so it's possible to query efficiently all the groups that a user is a member of. an index on group_id is not required since it's the first column of the primary key.
conversations table NOSQL
{
conversation_id,
user_id1,
user_id2
}
messages table NOSQL
{
pk: conversation_id,
sk: sequence_id,
}
acks table NOSQL
{
conversation_id
msg_id,
user_id,
}
presence table NOSQL
{
user_id,
websocket_server,
last_seen: timstamp
}
unread message queue table NOSQL
{
conversation_id,
msg_id,
receiver_id
}
The rest of these nosql tables will be stored in a lsm-tree db like cassandra or dynamodb. These DBs easily scale by the primary key of the table. For the conversations , messages, acks, and unread tables, the partition key will be the conversation_id. For the presence table, the partition key will be user_id.
load balancer - used to distribute messages accross servers.
api gateway - used to authenticate, authorize, and route requests.
Chat service - used to process user message requests.
presnece service - used to check whether user is online and to which websocket server it's connected to.
Kafka - event storage to separate message events from critical path.
Notification service - send receiver the message if he is online and store the message in a queue.
Send message flow:
receiver acknoledge message as read:
Databases:
I chosed to use NOSQL DBs like DyanomDB. That provides autoscaling to allow to handle growing user base and more requests load. The Users metadata will be stored on a sql DB for authentication and authorization and thus will require acid properties, but that will be easily stored on a single server with replication for fault tolerence.
The API Gateway component will route requests to the matching service according their path. This service also provides rate limiting in case of over loading of the service or if a user is abusing the service. The api gateway also provides authentication and authorization of requests.
The Chat service will be a stateless service and thus easily scalable. We can scale it by an average cpu metric and if it has surpassed the threshold, more pods will be scheduled to handle the traffic. To send notifications to the receiver, the chat service will send events to Kafka so it will not be part of the critical path of requests. Messages will be stored by creating first an auto incrementing sequence id using a sequencer service, that sequence id will be used to order messages according to the order of their arrival. The messages schema will use conversation_id as primary key and the sequence id as sort key. Thus, since the sort key is auto incremented, it will provide order to messages.
The notification service will pick up events from Kafka, and will query the presence service to see which receivers are online and to what websocket server they are connected to. The notification service will send the message to the websocket server that is connected to the receiver so it will send the message via websocket connection. The notificaion service will also send the message for the unread messages queue so the receiver could read the message in case it's offline at the time of sending the message.
Trading consistency for availability and low latency: in the system, I chose to use nosql DB to store messages with eventual consistency model and use kafka to store events and send them asynchrounsly to receivers. According to CAP model, we need to choose between consistency and availbility, and it's ok if not all users see immediately all the messages. We do care about having a system that responds quickly to users' request to improve user experience.
Websockets vs. long polling: I chose to use websockets instead of long polling since I favour low latency for receivers to get notified for messages. websockets are harder to scale since there is server affinity.
Delivery delays: in case of partial outages and spike in traffic, we use kafka as a buffer. Messages are queued in kafka until rest of the service is back and to send quick response to the client. If client doesn't get and ack for sending a message, the client will retry, and if not succeeding the api gateway will rate limit the client to prevent cascading failures. In this case, the client UI will receive a failure message to notify him that the delivery failed and that he should try again later.
The messages DB is sharded by using the conversation id as primary key. If a shard becomes to big, we can shard the older messages by using conversation_id#{partition_number} as primary key.
We can have a duplicate message delivery if the client doesn't receive and ack. We'll deal will this by the client sending an idempotence key for each post message request. Thus, the chat service will not store the same message twice.