Private messaging system (sharding)?

Question

MySQL

Hi all!
Can you please tell me how the system of personal messages between users in large projects (Vkontakte, Odnoklassniki, Topface, etc.) is arranged, taking into account scaling? It is the storage of data about users / chats / messages and access to this data that is of interest.
Let's say we have 100,000,000 users and we need to horizontally scale (shard) this data to 200 MySQL servers. At the moment, I see it as follows: we divide all users and data into 200 servers by user_id, we get about 500,000 users per server. You can also divide the data into spots of 1000 users and it turns out that each database server will have 500 spots of 1000 users (a total of 500,000 users per server).
Access to database servers/spots can be calculated by user_id, for example spot_id = user_id % 1000. Each spot will store data in the form of tables, for example:
Spot1:
- spot1_users (information about users)
- spot1_chats (information about chats between users)
- spot1_messages ( messages from chats)
....
Spot2:
- spot2_users
- spot2_chats
- spot2_messages
....
The problem arises when it is necessary to store/receive shared data between users. For example, 2 users start chatting with each other. In this case, you need to create a chat in the spotN_chats table and put the information chat_id (id of the chat), receiver_id (id of the recipient), sender_id (id of the sender) there. Messages will be stored in the messages table (chat_id, message, time).
Now the fun begins - users begin to correspond with each other. Here you need to do the following basic operations:
1) Creating a new chat between 2 users
2) Getting information about a chat or a list of chats for a specific user
3) Creating a new message
4) Getting a list of messages by chat_id
There are also 2 scenarios:
1) users are on the same spot (for example, spot1);
2) users are on different spots (for example, spot1 and spot2);
Task 1. User1 decided to start a chat with user2. In this case, you need to create a new chat in the database. If the users are on the same spot, then you can simply create a new chat in the spot1_chats table, get the chat_id, and then create new messages in the spot1_messages table with the received chat_id. But if users are on different spots (spot1, spot2), then this approach will not work, because in order for each user to see a list of their chats, they need to be duplicated on 2 spots at the same time. But in this case, chat_id will be different for 2 tables (spot1_chats, spot2_chats) if you use the autoincrement field for chat or you need to build some common id generator for 100 million users for new chats. Also, if 2 users are on the same spot,
Task 2. User1 sends a message to user2, we already have chat_id after creating a new chat. The same problem arises here as in the first problem. If 2 users are on the same spot, then we simply add a new message to the spot1_messages table, but if in the future we want to transfer the user to another spot, how to duplicate the messages? If users are at different spots, then to send a message, you need to create a new message in the spot1_messages table and in the spot2_messages table. In addition, if we want to update some new message counter or the time of the last message in chats, then we will also need to update the information about the chats in the spot1_chats and spot2_chats tables. It turns out that for a simple sending of one message, it will be necessary to make several queries to the database, namely:
Tell me, please, how to solve these problems correctly, or perhaps there is some other simpler / more reliable way to store messages from real experience.
Thank you.

Answer 1

https://vk.com/blog?id=653

Answer 2

Experience with Cassandra on Discord

Answer 3

As many as 2 architectures of the VKontakte private messaging system have been described - old and new How to rewrite the private messaging database from scratch...
5 trillion messages
2 billion messages per day

Answer 4

Do not try to do something mega-optimal right away, as they said above, it is difficult ... but you can get creative.
Firstly, store not user messages, but chats, respectively, and do sharding not by users, but by chats. Then the messages of one user will be spread across several servers at once but grouped into chats, just so that the chat itself (with any number of users, usually their number in the chat is much less than the total number of project users) uses only one server.
Secondly, separate operational and archived information, for chats this is a message box visible to users by default when opening a chat (you don’t download everything, it always happens in portions), but don’t make the transfer rule between these groups an axiom, let it be some a system process that moves data based on load and statistics. If users are limited to editing and deleting old messages, then it is possible to use different approaches to storing operational and archive information, different indexing technologies or even storage itself, incl. buffering optimizations... roughly speaking, you can store readonly archived data anywhere, because access to them does not require synchronization and locks.
It is not necessary to block a layer that gives information on where what is stored, for example, the location of archived chat messages can be stored in the same place as the operational one, it will still be requested, moreover, you need to try to do it everywhere so that it is immediately clear on which server the data is stored , the simplest way is a hash of the identifier (a simple remainder of dividing by the number of servers is more than adequate).
RAM data will generate the highest load, do your best to keep this data in RAM for as long as possible, i.e. the servers that process them must be configured differently from the archive ones.
ps mass mailings from the project or partners (don't make a face, they all want to do this) don't do it with regular chats, otherwise you will get strange quirks with mega-duplication of information and a huge load at the time of their implementation, let it be a separate entity (deal with the rest in the interface). On the other hand, if you need feedback, at the moment when the user suddenly decides to answer in such an underchat, convert it to a normal one.

Answer 5

found this option: sharding key for chat = sorted list of 2 user ids a
lot of useful things
highload.guide/blog
about sharding https://www.youtube.com/watch?v=URHoFbn4rt8
sharding in Badoo https://www.youtube. com/watch?v=ZGAHlGfW1yw
tech director topface on sharding and more https://spb-borodin.livejournal.com/
etsy sharding
https://www.slideshare.net/jgoulah/the-etsy-shard-.. https :
//www.slideshare.net/jgoulah/the-shard-revis...

Answer 6

1. It is possible to transfer users to a geo-averaged and less loaded message queue processing spot (IP address) dynamically (by geo-location of users, based on their IP).
2. Serve chat queues via Redis.
3. If the recipient(s) of the message is off-line, we drop the message directly into the database of off-line messages.