tdlight-team/tdlight
tdlight-team
/
tdlight
1
1
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Unify "log event" spelling. 

GitOrigin-RevId: a1a491981596c56dbf7541d80ead1ea9f5d5bd04
This commit is contained in:
levlam 2020-09-22 02:15:09 +03:00
parent 8503e7f214
commit cb3f390a92
31 changed files with 762 additions and 759 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
SplitSource.php
View File

@ -273,7 +273,7 @@ function split_file($file, $chunks, $undo) {
'HashtagHints' => 'HashtagHints', 'HashtagHints' => 'HashtagHints',
'inline_queries_manager[_(-][^.]|InlineQueriesManager' => 'InlineQueriesManager', 'inline_queries_manager[_(-][^.]|InlineQueriesManager' => 'InlineQueriesManager',
'language_pack_manager[_(-][^.]|LanguagePackManager' => 'LanguagePackManager', 'language_pack_manager[_(-][^.]|LanguagePackManager' => 'LanguagePackManager',
'LogeventIdWithGeneration|add_log_event|delete_log_event|get_erase_logevent_promise|parse_time|store_time' => 'logevent/LogEventHelper', 'LogeventIdWithGeneration|add_log_event|delete_log_event|get_erase_log_event_promise|parse_time|store_time' => 'logevent/LogEventHelper',
'MessageCopyOptions' => 'MessageCopyOptions', 'MessageCopyOptions' => 'MessageCopyOptions',
'messages_manager[_(-][^.]|MessagesManager' => 'MessagesManager', 'messages_manager[_(-][^.]|MessagesManager' => 'MessagesManager',
'notification_manager[_(-][^.]|NotificationManager|notifications[)]' => 'NotificationManager', 'notification_manager[_(-][^.]|NotificationManager|notifications[)]' => 'NotificationManager',

26
td/telegram/BackgroundManager.cpp
View File

@ -318,22 +318,22 @@ class BackgroundManager::BackgroundLogEvent {
void BackgroundManager::start_up() { void BackgroundManager::start_up() {
for (int i = 0; i < 2; i++) { for (int i = 0; i < 2; i++) {
bool for_dark_theme = i != 0; bool for_dark_theme = i != 0;
auto logevent_string = G()->td_db()->get_binlog_pmc()->get(get_background_database_key(for_dark_theme)); auto log_event_string = G()->td_db()->get_binlog_pmc()->get(get_background_database_key(for_dark_theme));
if (!logevent_string.empty()) { if (!log_event_string.empty()) {
BackgroundLogEvent logevent; BackgroundLogEvent log_event;
log_event_parse(logevent, logevent_string).ensure(); log_event_parse(log_event, log_event_string).ensure();
CHECK(logevent.background_.id.is_valid()); CHECK(log_event.background_.id.is_valid());
bool needs_file_id = logevent.background_.type.type != BackgroundType::Type::Fill; bool needs_file_id = log_event.background_.type.type != BackgroundType::Type::Fill;
if (logevent.background_.file_id.is_valid() != needs_file_id) { if (log_event.background_.file_id.is_valid() != needs_file_id) {
LOG(ERROR) << "Failed to load " << logevent.background_.id << " of " << logevent.background_.type; LOG(ERROR) << "Failed to load " << log_event.background_.id << " of " << log_event.background_.type;
G()->td_db()->get_binlog_pmc()->erase(get_background_database_key(for_dark_theme)); G()->td_db()->get_binlog_pmc()->erase(get_background_database_key(for_dark_theme));
continue; continue;
} }
set_background_id_[for_dark_theme] = logevent.background_.id; set_background_id_[for_dark_theme] = log_event.background_.id;
set_background_type_[for_dark_theme] = logevent.set_type_; set_background_type_[for_dark_theme] = log_event.set_type_;
add_background(logevent.background_); add_background(log_event.background_);
} }
send_update_selected_background(for_dark_theme); send_update_selected_background(for_dark_theme);
@ -662,8 +662,8 @@ void BackgroundManager::save_background_id(bool for_dark_theme) const {
if (background_id.is_valid()) { if (background_id.is_valid()) {
const Background *background = get_background(background_id); const Background *background = get_background(background_id);
CHECK(background != nullptr); CHECK(background != nullptr);
BackgroundLogEvent logevent{*background, set_background_type_[for_dark_theme]}; BackgroundLogEvent log_event{*background, set_background_type_[for_dark_theme]};
G()->td_db()->get_binlog_pmc()->set(key, log_event_store(logevent).as_slice().str()); G()->td_db()->get_binlog_pmc()->set(key, log_event_store(log_event).as_slice().str());
} else { } else {
G()->td_db()->get_binlog_pmc()->erase(key); G()->td_db()->get_binlog_pmc()->erase(key);
} }

104
td/telegram/ContactsManager.cpp
View File

@ -7313,12 +7313,12 @@ void ContactsManager::save_user(User *u, UserId user_id, bool from_binlog) {
CHECK(u != nullptr); CHECK(u != nullptr);
if (!u->is_saved || !u->is_status_saved) { // TODO more effective handling of !u->is_status_saved if (!u->is_saved || !u->is_status_saved) { // TODO more effective handling of !u->is_status_saved
if (!from_binlog) { if (!from_binlog) {
auto logevent = UserLogEvent(user_id, *u); auto log_event = UserLogEvent(user_id, *u);
auto storer = get_log_event_storer(logevent); auto storer = get_log_event_storer(log_event);
if (u->logevent_id == 0) { if (u->log_event_id == 0) {
u->logevent_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::Users, storer); u->log_event_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::Users, storer);
} else { } else {
binlog_rewrite(G()->td_db()->get_binlog(), u->logevent_id, LogEvent::HandlerType::Users, storer); binlog_rewrite(G()->td_db()->get_binlog(), u->log_event_id, LogEvent::HandlerType::Users, storer);
} }
} }
@ -7346,7 +7346,7 @@ void ContactsManager::on_binlog_user_event(BinlogEvent &&event) {
User *u = add_user(user_id, "on_binlog_user_event"); User *u = add_user(user_id, "on_binlog_user_event");
*u = std::move(log_event.u); // users come from binlog before all other events, so just add them *u = std::move(log_event.u); // users come from binlog before all other events, so just add them
u->logevent_id = event.id_; u->log_event_id = event.id_;
update_user(u, user_id, true, false); update_user(u, user_id, true, false);
} }
@ -7413,12 +7413,12 @@ void ContactsManager::on_save_user_to_database(UserId user_id, bool success) {
LOG(INFO) << "Successfully saved " << user_id << " to database"; LOG(INFO) << "Successfully saved " << user_id << " to database";
} }
if (u->is_saved && u->is_status_saved) { if (u->is_saved && u->is_status_saved) {
if (u->logevent_id != 0) { if (u->log_event_id != 0) {
binlog_erase(G()->td_db()->get_binlog(), u->logevent_id); binlog_erase(G()->td_db()->get_binlog(), u->log_event_id);
u->logevent_id = 0; u->log_event_id = 0;
} }
} else { } else {
save_user(u, user_id, u->logevent_id != 0); save_user(u, user_id, u->log_event_id != 0);
} }
} }
@ -7496,9 +7496,9 @@ void ContactsManager::on_load_user_from_database(UserId user_id, string value) {
auto new_value = get_user_database_value(u); auto new_value = get_user_database_value(u);
if (value != new_value) { if (value != new_value) {
save_user_to_database_impl(u, user_id, std::move(new_value)); save_user_to_database_impl(u, user_id, std::move(new_value));
} else if (u->logevent_id != 0) { } else if (u->log_event_id != 0) {
binlog_erase(G()->td_db()->get_binlog(), u->logevent_id); binlog_erase(G()->td_db()->get_binlog(), u->log_event_id);
u->logevent_id = 0; u->log_event_id = 0;
} }
} }
@ -7589,12 +7589,12 @@ void ContactsManager::save_chat(Chat *c, ChatId chat_id, bool from_binlog) {
CHECK(c != nullptr); CHECK(c != nullptr);
if (!c->is_saved) { if (!c->is_saved) {
if (!from_binlog) { if (!from_binlog) {
auto logevent = ChatLogEvent(chat_id, *c); auto log_event = ChatLogEvent(chat_id, *c);
auto storer = get_log_event_storer(logevent); auto storer = get_log_event_storer(log_event);
if (c->logevent_id == 0) { if (c->log_event_id == 0) {
c->logevent_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::Chats, storer); c->log_event_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::Chats, storer);
} else { } else {
binlog_rewrite(G()->td_db()->get_binlog(), c->logevent_id, LogEvent::HandlerType::Chats, storer); binlog_rewrite(G()->td_db()->get_binlog(), c->log_event_id, LogEvent::HandlerType::Chats, storer);
} }
} }
@ -7623,7 +7623,7 @@ void ContactsManager::on_binlog_chat_event(BinlogEvent &&event) {
Chat *c = add_chat(chat_id); Chat *c = add_chat(chat_id);
*c = std::move(log_event.c); // chats come from binlog before all other events, so just add them *c = std::move(log_event.c); // chats come from binlog before all other events, so just add them
c->logevent_id = event.id_; c->log_event_id = event.id_;
update_chat(c, chat_id, true, false); update_chat(c, chat_id, true, false);
} }
@ -7682,12 +7682,12 @@ void ContactsManager::on_save_chat_to_database(ChatId chat_id, bool success) {
LOG(INFO) << "Successfully saved " << chat_id << " to database"; LOG(INFO) << "Successfully saved " << chat_id << " to database";
} }
if (c->is_saved) { if (c->is_saved) {
if (c->logevent_id != 0) { if (c->log_event_id != 0) {
binlog_erase(G()->td_db()->get_binlog(), c->logevent_id); binlog_erase(G()->td_db()->get_binlog(), c->log_event_id);
c->logevent_id = 0; c->log_event_id = 0;
} }
} else { } else {
save_chat(c, chat_id, c->logevent_id != 0); save_chat(c, chat_id, c->log_event_id != 0);
} }
} }
@ -7751,9 +7751,9 @@ void ContactsManager::on_load_chat_from_database(ChatId chat_id, string value) {
auto new_value = get_chat_database_value(c); auto new_value = get_chat_database_value(c);
if (value != new_value) { if (value != new_value) {
save_chat_to_database_impl(c, chat_id, std::move(new_value)); save_chat_to_database_impl(c, chat_id, std::move(new_value));
} else if (c->logevent_id != 0) { } else if (c->log_event_id != 0) {
binlog_erase(G()->td_db()->get_binlog(), c->logevent_id); binlog_erase(G()->td_db()->get_binlog(), c->log_event_id);
c->logevent_id = 0; c->log_event_id = 0;
} }
} }
@ -7825,12 +7825,12 @@ void ContactsManager::save_channel(Channel *c, ChannelId channel_id, bool from_b
CHECK(c != nullptr); CHECK(c != nullptr);
if (!c->is_saved) { if (!c->is_saved) {
if (!from_binlog) { if (!from_binlog) {
auto logevent = ChannelLogEvent(channel_id, *c); auto log_event = ChannelLogEvent(channel_id, *c);
auto storer = get_log_event_storer(logevent); auto storer = get_log_event_storer(log_event);
if (c->logevent_id == 0) { if (c->log_event_id == 0) {
c->logevent_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::Channels, storer); c->log_event_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::Channels, storer);
} else { } else {
binlog_rewrite(G()->td_db()->get_binlog(), c->logevent_id, LogEvent::HandlerType::Channels, storer); binlog_rewrite(G()->td_db()->get_binlog(), c->log_event_id, LogEvent::HandlerType::Channels, storer);
} }
} }
@ -7859,7 +7859,7 @@ void ContactsManager::on_binlog_channel_event(BinlogEvent &&event) {
Channel *c = add_channel(channel_id, "on_binlog_channel_event"); Channel *c = add_channel(channel_id, "on_binlog_channel_event");
*c = std::move(log_event.c); // channels come from binlog before all other events, so just add them *c = std::move(log_event.c); // channels come from binlog before all other events, so just add them
c->logevent_id = event.id_; c->log_event_id = event.id_;
update_channel(c, channel_id, true, false); update_channel(c, channel_id, true, false);
} }
@ -7919,12 +7919,12 @@ void ContactsManager::on_save_channel_to_database(ChannelId channel_id, bool suc
LOG(INFO) << "Successfully saved " << channel_id << " to database"; LOG(INFO) << "Successfully saved " << channel_id << " to database";
} }
if (c->is_saved) { if (c->is_saved) {
if (c->logevent_id != 0) { if (c->log_event_id != 0) {
binlog_erase(G()->td_db()->get_binlog(), c->logevent_id); binlog_erase(G()->td_db()->get_binlog(), c->log_event_id);
c->logevent_id = 0; c->log_event_id = 0;
} }
} else { } else {
save_channel(c, channel_id, c->logevent_id != 0); save_channel(c, channel_id, c->log_event_id != 0);
} }
} }
@ -8008,9 +8008,9 @@ void ContactsManager::on_load_channel_from_database(ChannelId channel_id, string
auto new_value = get_channel_database_value(c); auto new_value = get_channel_database_value(c);
if (value != new_value) { if (value != new_value) {
save_channel_to_database_impl(c, channel_id, std::move(new_value)); save_channel_to_database_impl(c, channel_id, std::move(new_value));
} else if (c->logevent_id != 0) { } else if (c->log_event_id != 0) {
binlog_erase(G()->td_db()->get_binlog(), c->logevent_id); binlog_erase(G()->td_db()->get_binlog(), c->log_event_id);
c->logevent_id = 0; c->log_event_id = 0;
} }
} }
@ -8075,12 +8075,12 @@ void ContactsManager::save_secret_chat(SecretChat *c, SecretChatId secret_chat_i
CHECK(c != nullptr); CHECK(c != nullptr);
if (!c->is_saved) { if (!c->is_saved) {
if (!from_binlog) { if (!from_binlog) {
auto logevent = SecretChatLogEvent(secret_chat_id, *c); auto log_event = SecretChatLogEvent(secret_chat_id, *c);
auto storer = get_log_event_storer(logevent); auto storer = get_log_event_storer(log_event);
if (c->logevent_id == 0) { if (c->log_event_id == 0) {
c->logevent_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::SecretChatInfos, storer); c->log_event_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::SecretChatInfos, storer);
} else { } else {
binlog_rewrite(G()->td_db()->get_binlog(), c->logevent_id, LogEvent::HandlerType::SecretChatInfos, storer); binlog_rewrite(G()->td_db()->get_binlog(), c->log_event_id, LogEvent::HandlerType::SecretChatInfos, storer);
} }
} }
@ -8109,7 +8109,7 @@ void ContactsManager::on_binlog_secret_chat_event(BinlogEvent &&event) {
SecretChat *c = add_secret_chat(secret_chat_id); SecretChat *c = add_secret_chat(secret_chat_id);
*c = std::move(log_event.c); // secret chats come from binlog before all other events, so just add them *c = std::move(log_event.c); // secret chats come from binlog before all other events, so just add them
c->logevent_id = event.id_; c->log_event_id = event.id_;
update_secret_chat(c, secret_chat_id, true, false); update_secret_chat(c, secret_chat_id, true, false);
} }
@ -8170,12 +8170,12 @@ void ContactsManager::on_save_secret_chat_to_database(SecretChatId secret_chat_i
LOG(INFO) << "Successfully saved " << secret_chat_id << " to database"; LOG(INFO) << "Successfully saved " << secret_chat_id << " to database";
} }
if (c->is_saved) { if (c->is_saved) {
if (c->logevent_id != 0) { if (c->log_event_id != 0) {
binlog_erase(G()->td_db()->get_binlog(), c->logevent_id); binlog_erase(G()->td_db()->get_binlog(), c->log_event_id);
c->logevent_id = 0; c->log_event_id = 0;
} }
} else { } else {
save_secret_chat(c, secret_chat_id, c->logevent_id != 0); save_secret_chat(c, secret_chat_id, c->log_event_id != 0);
} }
} }
@ -8240,9 +8240,9 @@ void ContactsManager::on_load_secret_chat_from_database(SecretChatId secret_chat
auto new_value = get_secret_chat_database_value(c); auto new_value = get_secret_chat_database_value(c);
if (value != new_value) { if (value != new_value) {
save_secret_chat_to_database_impl(c, secret_chat_id, std::move(new_value)); save_secret_chat_to_database_impl(c, secret_chat_id, std::move(new_value));
} else if (c->logevent_id != 0) { } else if (c->log_event_id != 0) {
binlog_erase(G()->td_db()->get_binlog(), c->logevent_id); binlog_erase(G()->td_db()->get_binlog(), c->log_event_id);
c->logevent_id = 0; c->log_event_id = 0;
} }
} }

8
td/telegram/ContactsManager.h
View File

@ -630,7 +630,7 @@ class ContactsManager : public Actor {
bool is_received_from_server = false; // true, if the user was received from the server and not the database bool is_received_from_server = false; // true, if the user was received from the server and not the database
uint64 logevent_id = 0; uint64 log_event_id = 0;
template <class StorerT> template <class StorerT>
void store(StorerT &storer) const; void store(StorerT &storer) const;
@ -722,7 +722,7 @@ class ContactsManager : public Actor {
bool is_received_from_server = false; // true, if the chat was received from the server and not the database bool is_received_from_server = false; // true, if the chat was received from the server and not the database
uint64 logevent_id = 0; uint64 log_event_id = 0;
template <class StorerT> template <class StorerT>
void store(StorerT &storer) const; void store(StorerT &storer) const;
@ -799,7 +799,7 @@ class ContactsManager : public Actor {
bool is_received_from_server = false; // true, if the channel was received from the server and not the database bool is_received_from_server = false; // true, if the channel was received from the server and not the database
uint64 logevent_id = 0; uint64 log_event_id = 0;
template <class StorerT> template <class StorerT>
void store(StorerT &storer) const; void store(StorerT &storer) const;
@ -882,7 +882,7 @@ class ContactsManager : public Actor {
bool is_saved = false; // is current secret chat version being saved/is saved to the database bool is_saved = false; // is current secret chat version being saved/is saved to the database
bool is_being_saved = false; // is current secret chat being saved to the database bool is_being_saved = false; // is current secret chat being saved to the database
uint64 logevent_id = 0; uint64 log_event_id = 0;
template <class StorerT> template <class StorerT>
void store(StorerT &storer) const; void store(StorerT &storer) const;

579
td/telegram/MessagesManager.cpp
View File

File diff suppressed because it is too large Load Diff

100
td/telegram/MessagesManager.h
View File

@ -1093,7 +1093,7 @@ class MessagesManager : public Actor {
mutable int32 last_access_date = 0; mutable int32 last_access_date = 0;
mutable uint64 send_message_logevent_id = 0; mutable uint64 send_message_log_event_id = 0;
mutable NetQueryRef send_query_ref; mutable NetQueryRef send_query_ref;
@ -1146,11 +1146,11 @@ class MessagesManager : public Actor {
MessageId reply_markup_message_id; MessageId reply_markup_message_id;
DialogNotificationSettings notification_settings; DialogNotificationSettings notification_settings;
unique_ptr<DraftMessage> draft_message; unique_ptr<DraftMessage> draft_message;
LogeventIdWithGeneration save_draft_message_logevent_id; LogeventIdWithGeneration save_draft_message_log_event_id;
LogeventIdWithGeneration save_notification_settings_logevent_id; LogeventIdWithGeneration save_notification_settings_log_event_id;
std::unordered_map<int64, LogeventIdWithGeneration> read_history_logevent_ids; std::unordered_map<int64, LogeventIdWithGeneration> read_history_log_event_ids;
std::unordered_set<MessageId, MessageIdHash> updated_read_history_message_ids; std::unordered_set<MessageId, MessageIdHash> updated_read_history_message_ids;
LogeventIdWithGeneration set_folder_id_logevent_id; LogeventIdWithGeneration set_folder_id_log_event_id;
FolderId folder_id; FolderId folder_id;
vector<DialogListId> dialog_list_ids; // TODO replace with mask vector<DialogListId> dialog_list_ids; // TODO replace with mask
@ -1777,7 +1777,7 @@ class MessagesManager : public Actor {
bool skip_update_for_not_found_messages); bool skip_update_for_not_found_messages);
void do_forward_messages(DialogId to_dialog_id, DialogId from_dialog_id, const vector<Message *> &messages, void do_forward_messages(DialogId to_dialog_id, DialogId from_dialog_id, const vector<Message *> &messages,
const vector<MessageId> &message_ids, uint64 logevent_id); const vector<MessageId> &message_ids, uint64 log_event_id);
Result<MessageId> forward_message(DialogId to_dialog_id, DialogId from_dialog_id, MessageId message_id, Result<MessageId> forward_message(DialogId to_dialog_id, DialogId from_dialog_id, MessageId message_id,
tl_object_ptr<td_api::messageSendOptions> &&options, bool in_game_share, tl_object_ptr<td_api::messageSendOptions> &&options, bool in_game_share,
@ -1810,22 +1810,22 @@ class MessagesManager : public Actor {
void on_yet_unsent_media_queue_updated(DialogId dialog_id); void on_yet_unsent_media_queue_updated(DialogId dialog_id);
void save_send_bot_start_message_logevent(UserId bot_user_id, DialogId dialog_id, const string &parameter, void save_send_bot_start_message_log_event(UserId bot_user_id, DialogId dialog_id, const string &parameter,
const Message *m); const Message *m);
void do_send_bot_start_message(UserId bot_user_id, DialogId dialog_id, const string &parameter, const Message *m); void do_send_bot_start_message(UserId bot_user_id, DialogId dialog_id, const string &parameter, const Message *m);
void save_send_inline_query_result_message_logevent(DialogId dialog_id, const Message *m, int64 query_id, void save_send_inline_query_result_message_log_event(DialogId dialog_id, const Message *m, int64 query_id,
const string &result_id); const string &result_id);
void do_send_inline_query_result_message(DialogId dialog_id, const Message *m, int64 query_id, void do_send_inline_query_result_message(DialogId dialog_id, const Message *m, int64 query_id,
const string &result_id); const string &result_id);
uint64 save_send_screenshot_taken_notification_message_logevent(DialogId dialog_id, const Message *m); uint64 save_send_screenshot_taken_notification_message_log_event(DialogId dialog_id, const Message *m);
void do_send_screenshot_taken_notification_message(DialogId dialog_id, const Message *m, uint64 logevent_id); void do_send_screenshot_taken_notification_message(DialogId dialog_id, const Message *m, uint64 log_event_id);
Message *continue_send_message(DialogId dialog_id, unique_ptr<Message> &&m, uint64 logevent_id); Message *continue_send_message(DialogId dialog_id, unique_ptr<Message> &&m, uint64 log_event_id);
bool is_message_unload_enabled() const; bool is_message_unload_enabled() const;
@ -1867,19 +1867,19 @@ class MessagesManager : public Actor {
void delete_message_from_server(DialogId dialog_id, MessageId message_ids, bool revoke); void delete_message_from_server(DialogId dialog_id, MessageId message_ids, bool revoke);
void delete_messages_from_server(DialogId dialog_id, vector<MessageId> message_ids, bool revoke, uint64 logevent_id, void delete_messages_from_server(DialogId dialog_id, vector<MessageId> message_ids, bool revoke, uint64 log_event_id,
Promise<Unit> &&promise); Promise<Unit> &&promise);
void delete_scheduled_messages_from_server(DialogId dialog_id, vector<MessageId> message_ids, uint64 logevent_id, void delete_scheduled_messages_from_server(DialogId dialog_id, vector<MessageId> message_ids, uint64 log_event_id,
Promise<Unit> &&promise); Promise<Unit> &&promise);
void delete_dialog_history_from_server(DialogId dialog_id, MessageId max_message_id, bool remove_from_dialog_list, void delete_dialog_history_from_server(DialogId dialog_id, MessageId max_message_id, bool remove_from_dialog_list,
bool revoke, bool allow_error, uint64 logevent_id, Promise<Unit> &&promise); bool revoke, bool allow_error, uint64 log_event_id, Promise<Unit> &&promise);
void delete_all_channel_messages_from_user_on_server(ChannelId channel_id, UserId user_id, uint64 logevent_id, void delete_all_channel_messages_from_user_on_server(ChannelId channel_id, UserId user_id, uint64 log_event_id,
Promise<Unit> &&promise); Promise<Unit> &&promise);
void read_all_dialog_mentions_on_server(DialogId dialog_id, uint64 logevent_id, Promise<Unit> &&promise); void read_all_dialog_mentions_on_server(DialogId dialog_id, uint64 log_event_id, Promise<Unit> &&promise);
static MessageId find_message_by_date(const Message *m, int32 date); static MessageId find_message_by_date(const Message *m, int32 date);
@ -1918,8 +1918,8 @@ class MessagesManager : public Actor {
bool read_message_content(Dialog *d, Message *m, bool is_local_read, const char *source); bool read_message_content(Dialog *d, Message *m, bool is_local_read, const char *source);
void read_message_contents_on_server(DialogId dialog_id, vector<MessageId> message_ids, uint64 logevent_id, void read_message_contents_on_server(DialogId dialog_id, vector<MessageId> message_ids, uint64 log_event_id,
Promise<Unit> &&promise, bool skip_logevent = false); Promise<Unit> &&promise, bool skip_log_event = false);
bool has_incoming_notification(DialogId dialog_id, const Message *m) const; bool has_incoming_notification(DialogId dialog_id, const Message *m) const;
@ -2065,7 +2065,7 @@ class MessagesManager : public Actor {
NotificationId prev_last_notification_id, NotificationId prev_last_notification_id,
Result<vector<Notification>> result); Result<vector<Notification>> result);
void do_delete_message_logevent(const DeleteMessageLogEvent &logevent) const; void do_delete_message_log_event(const DeleteMessageLogEvent &log_event) const;
void attach_message_to_previous(Dialog *d, MessageId message_id, const char *source); void attach_message_to_previous(Dialog *d, MessageId message_id, const char *source);
@ -2265,13 +2265,13 @@ class MessagesManager : public Actor {
void do_set_dialog_folder_id(Dialog *d, FolderId folder_id); void do_set_dialog_folder_id(Dialog *d, FolderId folder_id);
void toggle_dialog_is_pinned_on_server(DialogId dialog_id, bool is_pinned, uint64 logevent_id); void toggle_dialog_is_pinned_on_server(DialogId dialog_id, bool is_pinned, uint64 log_event_id);
void toggle_dialog_is_marked_as_unread_on_server(DialogId dialog_id, bool is_marked_as_unread, uint64 logevent_id); void toggle_dialog_is_marked_as_unread_on_server(DialogId dialog_id, bool is_marked_as_unread, uint64 log_event_id);
void toggle_dialog_is_blocked_on_server(DialogId dialog_id, bool is_blocked, uint64 logevent_id); void toggle_dialog_is_blocked_on_server(DialogId dialog_id, bool is_blocked, uint64 log_event_id);
void reorder_pinned_dialogs_on_server(FolderId folder_id, const vector<DialogId> &dialog_ids, uint64 logevent_id); void reorder_pinned_dialogs_on_server(FolderId folder_id, const vector<DialogId> &dialog_ids, uint64 log_event_id);
void set_dialog_reply_markup(Dialog *d, MessageId message_id); void set_dialog_reply_markup(Dialog *d, MessageId message_id);
@ -2359,7 +2359,7 @@ class MessagesManager : public Actor {
void on_get_dialogs_from_database(FolderId folder_id, int32 limit, DialogDbGetDialogsResult &&dialogs, void on_get_dialogs_from_database(FolderId folder_id, int32 limit, DialogDbGetDialogsResult &&dialogs,
Promise<Unit> &&promise); Promise<Unit> &&promise);
void send_get_dialog_query(DialogId dialog_id, Promise<Unit> &&promise, uint64 logevent_id = 0); void send_get_dialog_query(DialogId dialog_id, Promise<Unit> &&promise, uint64 log_event_id = 0);
void send_search_public_dialogs_query(const string &query, Promise<Unit> &&promise); void send_search_public_dialogs_query(const string &query, Promise<Unit> &&promise);
@ -2645,11 +2645,11 @@ class MessagesManager : public Actor {
void on_updated_dialog_notification_settings(DialogId dialog_id, uint64 generation); void on_updated_dialog_notification_settings(DialogId dialog_id, uint64 generation);
void update_scope_notification_settings_on_server(NotificationSettingsScope scope, uint64 logevent_id); void update_scope_notification_settings_on_server(NotificationSettingsScope scope, uint64 log_event_id);
void reset_all_notification_settings_on_server(uint64 logevent_id); void reset_all_notification_settings_on_server(uint64 log_event_id);
void change_dialog_report_spam_state_on_server(DialogId dialog_id, bool is_spam_dialog, uint64 logevent_id, void change_dialog_report_spam_state_on_server(DialogId dialog_id, bool is_spam_dialog, uint64 log_event_id,
Promise<Unit> &&promise); Promise<Unit> &&promise);
void set_dialog_folder_id_on_server(DialogId dialog_id, bool from_binlog); void set_dialog_folder_id_on_server(DialogId dialog_id, bool from_binlog);
@ -2821,40 +2821,40 @@ class MessagesManager : public Actor {
static void add_message_dependencies(Dependencies &dependencies, DialogId dialog_id, const Message *m); static void add_message_dependencies(Dependencies &dependencies, DialogId dialog_id, const Message *m);
void save_send_message_logevent(DialogId dialog_id, const Message *m); void save_send_message_log_event(DialogId dialog_id, const Message *m);
uint64 save_change_dialog_report_spam_state_on_server_logevent(DialogId dialog_id, bool is_spam_dialog); uint64 save_change_dialog_report_spam_state_on_server_log_event(DialogId dialog_id, bool is_spam_dialog);
uint64 save_delete_messages_from_server_logevent(DialogId dialog_id, const vector<MessageId> &message_ids, uint64 save_delete_messages_from_server_log_event(DialogId dialog_id, const vector<MessageId> &message_ids,
bool revoke); bool revoke);
uint64 save_delete_scheduled_messages_from_server_logevent(DialogId dialog_id, const vector<MessageId> &message_ids); uint64 save_delete_scheduled_messages_from_server_log_event(DialogId dialog_id, const vector<MessageId> &message_ids);
uint64 save_delete_dialog_history_from_server_logevent(DialogId dialog_id, MessageId max_message_id, uint64 save_delete_dialog_history_from_server_log_event(DialogId dialog_id, MessageId max_message_id,
bool remove_from_dialog_list, bool revoke); bool remove_from_dialog_list, bool revoke);
uint64 save_delete_all_channel_messages_from_user_on_server_logevent(ChannelId channel_id, UserId user_id); uint64 save_delete_all_channel_messages_from_user_on_server_log_event(ChannelId channel_id, UserId user_id);
uint64 save_read_all_dialog_mentions_on_server_logevent(DialogId dialog_id); uint64 save_read_all_dialog_mentions_on_server_log_event(DialogId dialog_id);
uint64 save_toggle_dialog_is_pinned_on_server_logevent(DialogId dialog_id, bool is_pinned); uint64 save_toggle_dialog_is_pinned_on_server_log_event(DialogId dialog_id, bool is_pinned);
uint64 save_reorder_pinned_dialogs_on_server_logevent(FolderId folder_id, const vector<DialogId> &dialog_ids); uint64 save_reorder_pinned_dialogs_on_server_log_event(FolderId folder_id, const vector<DialogId> &dialog_ids);
uint64 save_toggle_dialog_is_marked_as_unread_on_server_logevent(DialogId dialog_id, bool is_marked_as_unread); uint64 save_toggle_dialog_is_marked_as_unread_on_server_log_event(DialogId dialog_id, bool is_marked_as_unread);
uint64 save_toggle_dialog_is_blocked_on_server_logevent(DialogId dialog_id, bool is_blocked); uint64 save_toggle_dialog_is_blocked_on_server_log_event(DialogId dialog_id, bool is_blocked);
uint64 save_read_message_contents_on_server_logevent(DialogId dialog_id, const vector<MessageId> &message_ids); uint64 save_read_message_contents_on_server_log_event(DialogId dialog_id, const vector<MessageId> &message_ids);
uint64 save_update_scope_notification_settings_on_server_logevent(NotificationSettingsScope scope); uint64 save_update_scope_notification_settings_on_server_log_event(NotificationSettingsScope scope);
uint64 save_reset_all_notification_settings_on_server_logevent(); uint64 save_reset_all_notification_settings_on_server_log_event();
uint64 save_get_dialog_from_server_logevent(DialogId dialog_id); uint64 save_get_dialog_from_server_log_event(DialogId dialog_id);
uint64 save_forward_messages_logevent(DialogId to_dialog_id, DialogId from_dialog_id, uint64 save_forward_messages_log_event(DialogId to_dialog_id, DialogId from_dialog_id,
const vector<Message *> &messages, const vector<MessageId> &message_ids); const vector<Message *> &messages, const vector<MessageId> &message_ids);
void suffix_load_loop(Dialog *d); void suffix_load_loop(Dialog *d);
void suffix_load_update_first_message_id(Dialog *d); void suffix_load_update_first_message_id(Dialog *d);
@ -3047,7 +3047,7 @@ class MessagesManager : public Actor {
std::unordered_map<DialogId, vector<Promise<Unit>>, DialogIdHash> get_dialog_notification_settings_queries_; std::unordered_map<DialogId, vector<Promise<Unit>>, DialogIdHash> get_dialog_notification_settings_queries_;
std::unordered_map<DialogId, vector<Promise<Unit>>, DialogIdHash> get_dialog_queries_; std::unordered_map<DialogId, vector<Promise<Unit>>, DialogIdHash> get_dialog_queries_;
std::unordered_map<DialogId, uint64, DialogIdHash> get_dialog_query_logevent_id_; std::unordered_map<DialogId, uint64, DialogIdHash> get_dialog_query_log_event_id_;
std::unordered_map<FullMessageId, int32, FullMessageIdHash> replied_by_yet_unsent_messages_; std::unordered_map<FullMessageId, int32, FullMessageIdHash> replied_by_yet_unsent_messages_;
@ -3093,7 +3093,7 @@ class MessagesManager : public Actor {
vector<Promise<Unit>> dialog_filter_reload_queries_; vector<Promise<Unit>> dialog_filter_reload_queries_;
std::unordered_map<DialogId, string, DialogIdHash> active_get_channel_differencies_; std::unordered_map<DialogId, string, DialogIdHash> active_get_channel_differencies_;
std::unordered_map<DialogId, uint64, DialogIdHash> get_channel_difference_to_logevent_id_; std::unordered_map<DialogId, uint64, DialogIdHash> get_channel_difference_to_log_event_id_;
MultiTimeout channel_get_difference_timeout_{"ChannelGetDifferenceTimeout"}; MultiTimeout channel_get_difference_timeout_{"ChannelGetDifferenceTimeout"};
MultiTimeout channel_get_difference_retry_timeout_{"ChannelGetDifferenceRetryTimeout"}; MultiTimeout channel_get_difference_retry_timeout_{"ChannelGetDifferenceRetryTimeout"};

82
td/telegram/NotificationManager.cpp
View File

@ -1654,25 +1654,25 @@ void NotificationManager::on_notification_processed(NotificationId notification_
void NotificationManager::on_notification_removed(NotificationId notification_id) { void NotificationManager::on_notification_removed(NotificationId notification_id) {
VLOG(notifications) << "In on_notification_removed with " << notification_id; VLOG(notifications) << "In on_notification_removed with " << notification_id;
auto add_it = temporary_notification_logevent_ids_.find(notification_id); auto add_it = temporary_notification_log_event_ids_.find(notification_id);
if (add_it == temporary_notification_logevent_ids_.end()) { if (add_it == temporary_notification_log_event_ids_.end()) {
return; return;
} }
auto edit_it = temporary_edit_notification_logevent_ids_.find(notification_id); auto edit_it = temporary_edit_notification_log_event_ids_.find(notification_id);
if (edit_it != temporary_edit_notification_logevent_ids_.end()) { if (edit_it != temporary_edit_notification_log_event_ids_.end()) {
VLOG(notifications) << "Remove from binlog edit of " << notification_id << " with logevent " << edit_it->second; VLOG(notifications) << "Remove from binlog edit of " << notification_id << " with log event " << edit_it->second;
if (!is_being_destroyed_) { if (!is_being_destroyed_) {
binlog_erase(G()->td_db()->get_binlog(), edit_it->second); binlog_erase(G()->td_db()->get_binlog(), edit_it->second);
} }
temporary_edit_notification_logevent_ids_.erase(edit_it); temporary_edit_notification_log_event_ids_.erase(edit_it);
} }
VLOG(notifications) << "Remove from binlog " << notification_id << " with logevent " << add_it->second; VLOG(notifications) << "Remove from binlog " << notification_id << " with log event " << add_it->second;
if (!is_being_destroyed_) { if (!is_being_destroyed_) {
binlog_erase(G()->td_db()->get_binlog(), add_it->second); binlog_erase(G()->td_db()->get_binlog(), add_it->second);
} }
temporary_notification_logevent_ids_.erase(add_it); temporary_notification_log_event_ids_.erase(add_it);
auto erased_notification_count = temporary_notifications_.erase(temporary_notification_message_ids_[notification_id]); auto erased_notification_count = temporary_notifications_.erase(temporary_notification_message_ids_[notification_id]);
auto erased_message_id_count = temporary_notification_message_ids_.erase(notification_id); auto erased_message_id_count = temporary_notification_message_ids_.erase(notification_id);
@ -3547,18 +3547,18 @@ void NotificationManager::add_message_push_notification(DialogId dialog_id, Mess
string sender_name, int32 date, bool is_from_scheduled, string sender_name, int32 date, bool is_from_scheduled,
bool contains_mention, bool initial_is_silent, bool is_silent, bool contains_mention, bool initial_is_silent, bool is_silent,
string loc_key, string arg, Photo photo, Document document, string loc_key, string arg, Photo photo, Document document,
NotificationId notification_id, uint64 logevent_id, NotificationId notification_id, uint64 log_event_id,
Promise<Unit> promise) { Promise<Unit> promise) {
auto is_pinned = begins_with(loc_key, "PINNED_"); auto is_pinned = begins_with(loc_key, "PINNED_");
auto r_info = td_->messages_manager_->get_message_push_notification_info( auto r_info = td_->messages_manager_->get_message_push_notification_info(
dialog_id, message_id, random_id, sender_user_id, sender_dialog_id, date, is_from_scheduled, contains_mention, dialog_id, message_id, random_id, sender_user_id, sender_dialog_id, date, is_from_scheduled, contains_mention,
is_pinned, logevent_id != 0); is_pinned, log_event_id != 0);
if (r_info.is_error()) { if (r_info.is_error()) {
VLOG(notifications) << "Don't need message push notification for " << message_id << "/" << random_id << " from " VLOG(notifications) << "Don't need message push notification for " << message_id << "/" << random_id << " from "
<< dialog_id << " sent by " << sender_user_id << "/" << sender_dialog_id << " at " << date << dialog_id << " sent by " << sender_user_id << "/" << sender_dialog_id << " at " << date
<< ": " << r_info.error(); << ": " << r_info.error();
if (logevent_id != 0) { if (log_event_id != 0) {
binlog_erase(G()->td_db()->get_binlog(), logevent_id); binlog_erase(G()->td_db()->get_binlog(), log_event_id);
} }
if (r_info.error().code() == 406) { if (r_info.error().code() == 406) {
promise.set_error(r_info.move_as_error()); promise.set_error(r_info.move_as_error());
@ -3576,24 +3576,24 @@ void NotificationManager::add_message_push_notification(DialogId dialog_id, Mess
// TODO support secret chat notifications // TODO support secret chat notifications
// main problem: there is no message_id yet // main problem: there is no message_id yet
// also don't forget to delete newSecretChat notification // also don't forget to delete newSecretChat notification
CHECK(logevent_id == 0); CHECK(log_event_id == 0);
return promise.set_error(Status::Error(406, "Secret chat push notifications are unsupported")); return promise.set_error(Status::Error(406, "Secret chat push notifications are unsupported"));
} }
CHECK(random_id == 0); CHECK(random_id == 0);
if (is_disabled() || max_notification_group_count_ == 0) { if (is_disabled() || max_notification_group_count_ == 0) {
CHECK(logevent_id == 0); CHECK(log_event_id == 0);
return promise.set_error(Status::Error(200, "Immediate success")); return promise.set_error(Status::Error(200, "Immediate success"));
} }
if (!notification_id.is_valid()) { if (!notification_id.is_valid()) {
CHECK(logevent_id == 0); CHECK(log_event_id == 0);
notification_id = get_next_notification_id(); notification_id = get_next_notification_id();
if (!notification_id.is_valid()) { if (!notification_id.is_valid()) {
return promise.set_value(Unit()); return promise.set_value(Unit());
} }
} else { } else {
CHECK(logevent_id != 0); CHECK(log_event_id != 0);
} }
if (sender_user_id.is_valid() && !td_->contacts_manager_->have_user_force(sender_user_id)) { if (sender_user_id.is_valid() && !td_->contacts_manager_->have_user_force(sender_user_id)) {
@ -3607,13 +3607,13 @@ void NotificationManager::add_message_push_notification(DialogId dialog_id, Mess
td_->contacts_manager_->on_get_user(std::move(user), "add_message_push_notification"); td_->contacts_manager_->on_get_user(std::move(user), "add_message_push_notification");
} }
if (logevent_id == 0 && G()->parameters().use_message_db) { if (log_event_id == 0 && G()->parameters().use_message_db) {
AddMessagePushNotificationLogEvent logevent{ AddMessagePushNotificationLogEvent log_event{
dialog_id, message_id, random_id, sender_user_id, sender_dialog_id, sender_name, dialog_id, message_id, random_id, sender_user_id, sender_dialog_id, sender_name,
date, is_from_scheduled, contains_mention, initial_is_silent, loc_key, arg, date, is_from_scheduled, contains_mention, initial_is_silent, loc_key, arg,
photo, document, notification_id}; photo, document, notification_id};
logevent_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::AddMessagePushNotification, log_event_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::AddMessagePushNotification,
get_log_event_storer(logevent)); get_log_event_storer(log_event));
} }
auto group_id = info.group_id; auto group_id = info.group_id;
@ -3621,9 +3621,9 @@ void NotificationManager::add_message_push_notification(DialogId dialog_id, Mess
bool is_outgoing = bool is_outgoing =
sender_user_id.is_valid() ? td_->contacts_manager_->get_my_id() == sender_user_id : is_from_scheduled; sender_user_id.is_valid() ? td_->contacts_manager_->get_my_id() == sender_user_id : is_from_scheduled;
if (logevent_id != 0) { if (log_event_id != 0) {
VLOG(notifications) << "Register temporary " << notification_id << " with logevent " << logevent_id; VLOG(notifications) << "Register temporary " << notification_id << " with log event " << log_event_id;
temporary_notification_logevent_ids_[notification_id] = logevent_id; temporary_notification_log_event_ids_[notification_id] = log_event_id;
temporary_notifications_[FullMessageId(dialog_id, message_id)] = {group_id, notification_id, sender_user_id, temporary_notifications_[FullMessageId(dialog_id, message_id)] = {group_id, notification_id, sender_user_id,
sender_dialog_id, sender_name, is_outgoing}; sender_dialog_id, sender_name, is_outgoing};
temporary_notification_message_ids_[notification_id] = FullMessageId(dialog_id, message_id); temporary_notification_message_ids_[notification_id] = FullMessageId(dialog_id, message_id);
@ -3716,9 +3716,9 @@ class NotificationManager::EditMessagePushNotificationLogEvent {
void NotificationManager::edit_message_push_notification(DialogId dialog_id, MessageId message_id, int32 edit_date, void NotificationManager::edit_message_push_notification(DialogId dialog_id, MessageId message_id, int32 edit_date,
string loc_key, string arg, Photo photo, Document document, string loc_key, string arg, Photo photo, Document document,
uint64 logevent_id, Promise<Unit> promise) { uint64 log_event_id, Promise<Unit> promise) {
if (is_disabled() || max_notification_group_count_ == 0) { if (is_disabled() || max_notification_group_count_ == 0) {
CHECK(logevent_id == 0); CHECK(log_event_id == 0);
return promise.set_error(Status::Error(200, "Immediate success")); return promise.set_error(Status::Error(200, "Immediate success"));
} }
@ -3738,23 +3738,23 @@ void NotificationManager::edit_message_push_notification(DialogId dialog_id, Mes
CHECK(group_id.is_valid()); CHECK(group_id.is_valid());
CHECK(notification_id.is_valid()); CHECK(notification_id.is_valid());
if (logevent_id == 0 && G()->parameters().use_message_db) { if (log_event_id == 0 && G()->parameters().use_message_db) {
EditMessagePushNotificationLogEvent logevent{dialog_id, message_id, edit_date, loc_key, arg, photo, document}; EditMessagePushNotificationLogEvent log_event{dialog_id, message_id, edit_date, loc_key, arg, photo, document};
auto storer = get_log_event_storer(logevent); auto storer = get_log_event_storer(log_event);
auto &cur_logevent_id = temporary_edit_notification_logevent_ids_[notification_id]; auto &cur_log_event_id = temporary_edit_notification_log_event_ids_[notification_id];
if (cur_logevent_id == 0) { if (cur_log_event_id == 0) {
logevent_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::EditMessagePushNotification, storer); log_event_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::EditMessagePushNotification, storer);
cur_logevent_id = logevent_id; cur_log_event_id = log_event_id;
VLOG(notifications) << "Add edit message push notification logevent " << logevent_id; VLOG(notifications) << "Add edit message push notification log event " << log_event_id;
} else { } else {
auto new_logevent_id = binlog_rewrite(G()->td_db()->get_binlog(), cur_logevent_id, auto new_log_event_id = binlog_rewrite(G()->td_db()->get_binlog(), cur_log_event_id,
LogEvent::HandlerType::EditMessagePushNotification, storer); LogEvent::HandlerType::EditMessagePushNotification, storer);
VLOG(notifications) << "Rewrite edit message push notification logevent " << cur_logevent_id << " with " VLOG(notifications) << "Rewrite edit message push notification log event " << cur_log_event_id << " with "
<< new_logevent_id; << new_log_event_id;
} }
} else if (logevent_id != 0) { } else if (log_event_id != 0) {
VLOG(notifications) << "Register edit of temporary " << notification_id << " with logevent " << logevent_id; VLOG(notifications) << "Register edit of temporary " << notification_id << " with log event " << log_event_id;
temporary_edit_notification_logevent_ids_[notification_id] = logevent_id; temporary_edit_notification_log_event_ids_[notification_id] = log_event_id;
} }
push_notification_promises_[notification_id].push_back(std::move(promise)); push_notification_promises_[notification_id].push_back(std::move(promise));
@ -4113,7 +4113,7 @@ void NotificationManager::on_binlog_events(vector<BinlogEvent> &&events) {
break; break;
} }
default: default:
LOG(FATAL) << "Unsupported logevent type " << event.type_; LOG(FATAL) << "Unsupported log event type " << event.type_;
} }
} }
if (is_inited_) { if (is_inited_) {

8
td/telegram/NotificationManager.h
View File

@ -310,10 +310,10 @@ class NotificationManager : public Actor {
DialogId sender_dialog_id, string sender_name, int32 date, bool is_from_scheduled, DialogId sender_dialog_id, string sender_name, int32 date, bool is_from_scheduled,
bool contains_mention, bool initial_is_silent, bool is_silent, string loc_key, bool contains_mention, bool initial_is_silent, bool is_silent, string loc_key,
string arg, Photo photo, Document document, NotificationId notification_id, string arg, Photo photo, Document document, NotificationId notification_id,
uint64 logevent_id, Promise<Unit> promise); uint64 log_event_id, Promise<Unit> promise);
void edit_message_push_notification(DialogId dialog_id, MessageId message_id, int32 edit_date, string loc_key, void edit_message_push_notification(DialogId dialog_id, MessageId message_id, int32 edit_date, string loc_key,
string arg, Photo photo, Document document, uint64 logevent_id, string arg, Photo photo, Document document, uint64 log_event_id,
Promise<Unit> promise); Promise<Unit> promise);
void after_get_difference_impl(); void after_get_difference_impl();
@ -382,8 +382,8 @@ class NotificationManager : public Actor {
std::unordered_set<NotificationGroupId, NotificationGroupIdHash> available_call_notification_group_ids_; std::unordered_set<NotificationGroupId, NotificationGroupIdHash> available_call_notification_group_ids_;
std::unordered_map<DialogId, NotificationGroupId, DialogIdHash> dialog_id_to_call_notification_group_id_; std::unordered_map<DialogId, NotificationGroupId, DialogIdHash> dialog_id_to_call_notification_group_id_;
std::unordered_map<NotificationId, uint64, NotificationIdHash> temporary_notification_logevent_ids_; std::unordered_map<NotificationId, uint64, NotificationIdHash> temporary_notification_log_event_ids_;
std::unordered_map<NotificationId, uint64, NotificationIdHash> temporary_edit_notification_logevent_ids_; std::unordered_map<NotificationId, uint64, NotificationIdHash> temporary_edit_notification_log_event_ids_;
struct TemporaryNotification { struct TemporaryNotification {
NotificationGroupId group_id; NotificationGroupId group_id;
NotificationId notification_id; NotificationId notification_id;

58
td/telegram/PollManager.cpp
View File

@ -749,7 +749,7 @@ class PollManager::SetPollAnswerLogEvent {
}; };
void PollManager::do_set_poll_answer(PollId poll_id, FullMessageId full_message_id, vector<string> &&options, void PollManager::do_set_poll_answer(PollId poll_id, FullMessageId full_message_id, vector<string> &&options,
uint64 logevent_id, Promise<Unit> &&promise) { uint64 log_event_id, Promise<Unit> &&promise) {
LOG(INFO) << "Set answer in " << poll_id << " from " << full_message_id; LOG(INFO) << "Set answer in " << poll_id << " from " << full_message_id;
auto &pending_answer = pending_answers_[poll_id]; auto &pending_answer = pending_answers_[poll_id];
if (!pending_answer.promises_.empty() && pending_answer.options_ == options) { if (!pending_answer.promises_.empty() && pending_answer.options_ == options) {
@ -757,27 +757,27 @@ void PollManager::do_set_poll_answer(PollId poll_id, FullMessageId full_message_
return; return;
} }
if (pending_answer.logevent_id_ != 0 && logevent_id != 0) { if (pending_answer.log_event_id_ != 0 && log_event_id != 0) {
LOG(ERROR) << "Duplicate SetPollAnswer log event: " << pending_answer.logevent_id_ << " and " << logevent_id; LOG(ERROR) << "Duplicate SetPollAnswer log event: " << pending_answer.log_event_id_ << " and " << log_event_id;
binlog_erase(G()->td_db()->get_binlog(), logevent_id); binlog_erase(G()->td_db()->get_binlog(), log_event_id);
return; return;
} }
if (logevent_id == 0 && G()->parameters().use_message_db) { if (log_event_id == 0 && G()->parameters().use_message_db) {
SetPollAnswerLogEvent logevent; SetPollAnswerLogEvent log_event;
logevent.poll_id_ = poll_id; log_event.poll_id_ = poll_id;
logevent.full_message_id_ = full_message_id; log_event.full_message_id_ = full_message_id;
logevent.options_ = options; log_event.options_ = options;
auto storer = get_log_event_storer(logevent); auto storer = get_log_event_storer(log_event);
if (pending_answer.generation_ == 0) { if (pending_answer.generation_ == 0) {
CHECK(pending_answer.logevent_id_ == 0); CHECK(pending_answer.log_event_id_ == 0);
logevent_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::SetPollAnswer, storer); log_event_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::SetPollAnswer, storer);
LOG(INFO) << "Add set poll answer logevent " << logevent_id; LOG(INFO) << "Add set poll answer log event " << log_event_id;
} else { } else {
CHECK(pending_answer.logevent_id_ != 0); CHECK(pending_answer.log_event_id_ != 0);
logevent_id = pending_answer.logevent_id_; log_event_id = pending_answer.log_event_id_;
auto new_logevent_id = binlog_rewrite(G()->td_db()->get_binlog(), pending_answer.logevent_id_, auto new_log_event_id = binlog_rewrite(G()->td_db()->get_binlog(), pending_answer.log_event_id_,
LogEvent::HandlerType::SetPollAnswer, storer); LogEvent::HandlerType::SetPollAnswer, storer);
LOG(INFO) << "Rewrite set poll answer logevent " << logevent_id << " with " << new_logevent_id; LOG(INFO) << "Rewrite set poll answer log event " << log_event_id << " with " << new_log_event_id;
} }
} }
@ -803,7 +803,7 @@ void PollManager::do_set_poll_answer(PollId poll_id, FullMessageId full_message_
pending_answer.options_ = std::move(options); pending_answer.options_ = std::move(options);
pending_answer.promises_.push_back(std::move(promise)); pending_answer.promises_.push_back(std::move(promise));
pending_answer.generation_ = generation; pending_answer.generation_ = generation;
pending_answer.logevent_id_ = logevent_id; pending_answer.log_event_id_ = log_event_id;
notify_on_poll_update(poll_id); notify_on_poll_update(poll_id);
@ -833,9 +833,9 @@ void PollManager::on_set_poll_answer(PollId poll_id, uint64 generation,
return; return;
} }
if (pending_answer.logevent_id_ != 0) { if (pending_answer.log_event_id_ != 0) {
LOG(INFO) << "Delete set poll answer logevent " << pending_answer.logevent_id_; LOG(INFO) << "Delete set poll answer log event " << pending_answer.log_event_id_;
binlog_erase(G()->td_db()->get_binlog(), pending_answer.logevent_id_); binlog_erase(G()->td_db()->get_binlog(), pending_answer.log_event_id_);
} }
auto promises = std::move(pending_answer.promises_); auto promises = std::move(pending_answer.promises_);
@ -1104,17 +1104,17 @@ class PollManager::StopPollLogEvent {
}; };
void PollManager::do_stop_poll(PollId poll_id, FullMessageId full_message_id, unique_ptr<ReplyMarkup> &&reply_markup, void PollManager::do_stop_poll(PollId poll_id, FullMessageId full_message_id, unique_ptr<ReplyMarkup> &&reply_markup,
uint64 logevent_id, Promise<Unit> &&promise) { uint64 log_event_id, Promise<Unit> &&promise) {
LOG(INFO) << "Stop " << poll_id << " from " << full_message_id; LOG(INFO) << "Stop " << poll_id << " from " << full_message_id;
if (logevent_id == 0 && G()->parameters().use_message_db && reply_markup == nullptr) { if (log_event_id == 0 && G()->parameters().use_message_db && reply_markup == nullptr) {
StopPollLogEvent logevent{poll_id, full_message_id}; StopPollLogEvent log_event{poll_id, full_message_id};
logevent_id = log_event_id =
binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::StopPoll, get_log_event_storer(logevent)); binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::StopPoll, get_log_event_storer(log_event));
} }
bool is_inserted = being_closed_polls_.insert(poll_id).second; bool is_inserted = being_closed_polls_.insert(poll_id).second;
CHECK(is_inserted); CHECK(is_inserted);
auto new_promise = get_erase_logevent_promise(logevent_id, std::move(promise)); auto new_promise = get_erase_log_event_promise(log_event_id, std::move(promise));
send_closure(td_->create_net_actor<StopPollActor>(std::move(new_promise)), &StopPollActor::send, full_message_id, send_closure(td_->create_net_actor<StopPollActor>(std::move(new_promise)), &StopPollActor::send, full_message_id,
std::move(reply_markup)); std::move(reply_markup));
@ -1678,7 +1678,7 @@ void PollManager::on_binlog_events(vector<BinlogEvent> &&events) {
break; break;
} }
default: default:
LOG(FATAL) << "Unsupported logevent type " << event.type_; LOG(FATAL) << "Unsupported log event type " << event.type_;
} }
} }
} }

6
td/telegram/PollManager.h
View File

@ -181,7 +181,7 @@ class PollManager : public Actor {
void on_get_poll_results(PollId poll_id, uint64 generation, Result<tl_object_ptr<telegram_api::Updates>> result); void on_get_poll_results(PollId poll_id, uint64 generation, Result<tl_object_ptr<telegram_api::Updates>> result);
void do_set_poll_answer(PollId poll_id, FullMessageId full_message_id, vector<string> &&options, uint64 logevent_id, void do_set_poll_answer(PollId poll_id, FullMessageId full_message_id, vector<string> &&options, uint64 log_event_id,
Promise<Unit> &&promise); Promise<Unit> &&promise);
void on_set_poll_answer(PollId poll_id, uint64 generation, Result<tl_object_ptr<telegram_api::Updates>> &&result); void on_set_poll_answer(PollId poll_id, uint64 generation, Result<tl_object_ptr<telegram_api::Updates>> &&result);
@ -196,7 +196,7 @@ class PollManager : public Actor {
Result<tl_object_ptr<telegram_api::messages_votesList>> &&result); Result<tl_object_ptr<telegram_api::messages_votesList>> &&result);
void do_stop_poll(PollId poll_id, FullMessageId full_message_id, unique_ptr<ReplyMarkup> &&reply_markup, void do_stop_poll(PollId poll_id, FullMessageId full_message_id, unique_ptr<ReplyMarkup> &&reply_markup,
uint64 logevent_id, Promise<Unit> &&promise); uint64 log_event_id, Promise<Unit> &&promise);
MultiTimeout update_poll_timeout_{"UpdatePollTimeout"}; MultiTimeout update_poll_timeout_{"UpdatePollTimeout"};
MultiTimeout close_poll_timeout_{"ClosePollTimeout"}; MultiTimeout close_poll_timeout_{"ClosePollTimeout"};
@ -211,7 +211,7 @@ class PollManager : public Actor {
vector<string> options_; vector<string> options_;
vector<Promise<Unit>> promises_; vector<Promise<Unit>> promises_;
uint64 generation_ = 0; uint64 generation_ = 0;
uint64 logevent_id_ = 0; uint64 log_event_id_ = 0;
NetQueryRef query_ref_; NetQueryRef query_ref_;
}; };
std::unordered_map<PollId, PendingPollAnswer, PollIdHash> pending_answers_; std::unordered_map<PollId, PendingPollAnswer, PollIdHash> pending_answers_;

226
td/telegram/SecretChatActor.cpp
View File

@ -96,11 +96,11 @@ void SecretChatActor::create_chat(int32 user_id, int64 user_access_hash, int32 r
return; return;
} }
auto event = make_unique<logevent::CreateSecretChat>(); auto event = make_unique<log_event::CreateSecretChat>();
event->user_id = user_id; event->user_id = user_id;
event->user_access_hash = user_access_hash; event->user_access_hash = user_access_hash;
event->random_id = random_id; event->random_id = random_id;
event->set_logevent_id(binlog_add(context_->binlog(), LogEvent::HandlerType::SecretChats, create_storer(*event))); event->set_log_event_id(binlog_add(context_->binlog(), LogEvent::HandlerType::SecretChats, create_storer(*event)));
do_create_chat_impl(std::move(event)); do_create_chat_impl(std::move(event));
promise.set_value(SecretChatId(random_id)); promise.set_value(SecretChatId(random_id));
loop(); loop();
@ -139,18 +139,18 @@ void SecretChatActor::on_result_resendable(NetQueryPtr net_query, Promise<NetQue
loop(); loop();
} }
void SecretChatActor::replay_close_chat(unique_ptr<logevent::CloseSecretChat> event) { void SecretChatActor::replay_close_chat(unique_ptr<log_event::CloseSecretChat> event) {
do_close_chat_impl(std::move(event)); do_close_chat_impl(std::move(event));
} }
void SecretChatActor::replay_create_chat(unique_ptr<logevent::CreateSecretChat> event) { void SecretChatActor::replay_create_chat(unique_ptr<log_event::CreateSecretChat> event) {
if (close_flag_) { if (close_flag_) {
return; return;
} }
do_create_chat_impl(std::move(event)); do_create_chat_impl(std::move(event));
} }
void SecretChatActor::add_inbound_message(unique_ptr<logevent::InboundSecretMessage> message) { void SecretChatActor::add_inbound_message(unique_ptr<log_event::InboundSecretMessage> message) {
SCOPE_EXIT { SCOPE_EXIT {
if (message) { if (message) {
message->promise.set_value(Unit()); message->promise.set_value(Unit());
@ -167,7 +167,7 @@ void SecretChatActor::add_inbound_message(unique_ptr<logevent::InboundSecretMess
loop(); loop();
} }
void SecretChatActor::replay_inbound_message(unique_ptr<logevent::InboundSecretMessage> message) { void SecretChatActor::replay_inbound_message(unique_ptr<log_event::InboundSecretMessage> message) {
if (close_flag_) { if (close_flag_) {
return; return;
} }
@ -190,7 +190,7 @@ void SecretChatActor::replay_inbound_message(unique_ptr<logevent::InboundSecretM
loop(); loop();
} }
void SecretChatActor::replay_outbound_message(unique_ptr<logevent::OutboundSecretMessage> message) { void SecretChatActor::replay_outbound_message(unique_ptr<log_event::OutboundSecretMessage> message) {
if (close_flag_) { if (close_flag_) {
return; return;
} }
@ -331,10 +331,10 @@ void SecretChatActor::send_message_impl(tl_object_ptr<secret_api::DecryptedMessa
return on_outbound_outer_send_message_promise(it->second, std::move(promise)); return on_outbound_outer_send_message_promise(it->second, std::move(promise));
} }
auto binlog_event = make_unique<logevent::OutboundSecretMessage>(); auto binlog_event = make_unique<log_event::OutboundSecretMessage>();
binlog_event->chat_id = auth_state_.id; binlog_event->chat_id = auth_state_.id;
binlog_event->random_id = random_id; binlog_event->random_id = random_id;
binlog_event->file = logevent::EncryptedInputFile::from_input_encrypted_file(file); binlog_event->file = log_event::EncryptedInputFile::from_input_encrypted_file(file);
binlog_event->message_id = seq_no_state_.message_id + 1; binlog_event->message_id = seq_no_state_.message_id + 1;
binlog_event->my_in_seq_no = seq_no_state_.my_in_seq_no; binlog_event->my_in_seq_no = seq_no_state_.my_in_seq_no;
binlog_event->my_out_seq_no = seq_no_state_.my_out_seq_no + 1; binlog_event->my_out_seq_no = seq_no_state_.my_out_seq_no + 1;
@ -719,23 +719,23 @@ void SecretChatActor::cancel_chat(Promise<> promise) {
} }
close_flag_ = true; close_flag_ = true;
std::vector<logevent::LogEvent::Id> to_delete; std::vector<log_event::LogEvent::Id> to_delete;
outbound_message_states_.for_each( outbound_message_states_.for_each(
[&](auto state_id, auto &state) { to_delete.push_back(state.message->logevent_id()); }); [&](auto state_id, auto &state) { to_delete.push_back(state.message->log_event_id()); });
inbound_message_states_.for_each([&](auto state_id, auto &state) { to_delete.push_back(state.logevent_id); }); inbound_message_states_.for_each([&](auto state_id, auto &state) { to_delete.push_back(state.log_event_id); });
// TODO: It must be a transaction // TODO: It must be a transaction
for (auto id : to_delete) { for (auto id : to_delete) {
binlog_erase(context_->binlog(), id); binlog_erase(context_->binlog(), id);
} }
if (create_logevent_id_ != 0) { if (create_log_event_id_ != 0) {
binlog_erase(context_->binlog(), create_logevent_id_); binlog_erase(context_->binlog(), create_log_event_id_);
create_logevent_id_ = 0; create_log_event_id_ = 0;
} }
auto event = make_unique<logevent::CloseSecretChat>(); auto event = make_unique<log_event::CloseSecretChat>();
event->chat_id = auth_state_.id; event->chat_id = auth_state_.id;
event->set_logevent_id(binlog_add(context_->binlog(), LogEvent::HandlerType::SecretChats, create_storer(*event))); event->set_log_event_id(binlog_add(context_->binlog(), LogEvent::HandlerType::SecretChats, create_storer(*event)));
auto on_sync = PromiseCreator::lambda( auto on_sync = PromiseCreator::lambda(
[actor_id = actor_id(this), event = std::move(event), promise = std::move(promise)](Result<Unit> result) mutable { [actor_id = actor_id(this), event = std::move(event), promise = std::move(promise)](Result<Unit> result) mutable {
@ -752,9 +752,9 @@ void SecretChatActor::cancel_chat(Promise<> promise) {
yield(); yield();
} }
void SecretChatActor::do_close_chat_impl(unique_ptr<logevent::CloseSecretChat> event) { void SecretChatActor::do_close_chat_impl(unique_ptr<log_event::CloseSecretChat> event) {
close_flag_ = true; close_flag_ = true;
close_logevent_id_ = event->logevent_id(); close_log_event_id_ = event->log_event_id();
LOG(INFO) << "Send messages.discardEncryption"; LOG(INFO) << "Send messages.discardEncryption";
auth_state_.state = State::Closed; auth_state_.state = State::Closed;
context_->secret_chat_db()->set_value(auth_state_); context_->secret_chat_db()->set_value(auth_state_);
@ -768,10 +768,10 @@ void SecretChatActor::do_close_chat_impl(unique_ptr<logevent::CloseSecretChat> e
context_->send_net_query(std::move(query), actor_shared(this), true); context_->send_net_query(std::move(query), actor_shared(this), true);
} }
void SecretChatActor::do_create_chat_impl(unique_ptr<logevent::CreateSecretChat> event) { void SecretChatActor::do_create_chat_impl(unique_ptr<log_event::CreateSecretChat> event) {
LOG(INFO) << *event; LOG(INFO) << *event;
CHECK(event->random_id == auth_state_.id); CHECK(event->random_id == auth_state_.id);
create_logevent_id_ = event->logevent_id(); create_log_event_id_ = event->log_event_id();
if (auth_state_.state == State::Empty) { if (auth_state_.state == State::Empty) {
auth_state_.user_id = event->user_id; auth_state_.user_id = event->user_id;
@ -784,19 +784,19 @@ void SecretChatActor::do_create_chat_impl(unique_ptr<logevent::CreateSecretChat>
} else if (auth_state_.state == State::SendRequest) { } else if (auth_state_.state == State::SendRequest) {
} else if (auth_state_.state == State::WaitRequestResponse) { } else if (auth_state_.state == State::WaitRequestResponse) {
} else { } else {
binlog_erase(context_->binlog(), create_logevent_id_); binlog_erase(context_->binlog(), create_log_event_id_);
create_logevent_id_ = 0; create_log_event_id_ = 0;
} }
} }
void SecretChatActor::on_discard_encryption_result(NetQueryPtr result) { void SecretChatActor::on_discard_encryption_result(NetQueryPtr result) {
CHECK(close_flag_); CHECK(close_flag_);
CHECK(close_logevent_id_ != 0); CHECK(close_log_event_id_ != 0);
if (context_->close_flag()) { if (context_->close_flag()) {
return; return;
} }
LOG(INFO) << "Got result for messages.discardEncryption"; LOG(INFO) << "Got result for messages.discardEncryption";
context_->secret_chat_db()->erase_value(auth_state_); context_->secret_chat_db()->erase_value(auth_state_);
binlog_erase(context_->binlog(), close_logevent_id_); binlog_erase(context_->binlog(), close_log_event_id_);
// skip flush // skip flush
stop(); stop();
} }
@ -877,7 +877,7 @@ Result<std::tuple<uint64, BufferSlice, int32>> SecretChatActor::decrypt(BufferSl
} }
} }
Status SecretChatActor::do_inbound_message_encrypted(unique_ptr<logevent::InboundSecretMessage> message) { Status SecretChatActor::do_inbound_message_encrypted(unique_ptr<log_event::InboundSecretMessage> message) {
SCOPE_EXIT { SCOPE_EXIT {
if (message) { if (message) {
message->promise.set_value(Unit()); message->promise.set_value(Unit());
@ -970,7 +970,7 @@ Status SecretChatActor::check_seq_no(int in_seq_no, int out_seq_no, int32 his_la
return Status::OK(); return Status::OK();
} }
Status SecretChatActor::do_inbound_message_decrypted_unchecked(unique_ptr<logevent::InboundSecretMessage> message) { Status SecretChatActor::do_inbound_message_decrypted_unchecked(unique_ptr<log_event::InboundSecretMessage> message) {
SCOPE_EXIT { SCOPE_EXIT {
CHECK(message == nullptr || !message->promise); CHECK(message == nullptr || !message->promise);
}; };
@ -979,9 +979,9 @@ Status SecretChatActor::do_inbound_message_decrypted_unchecked(unique_ptr<logeve
auto status = check_seq_no(in_seq_no, out_seq_no, message->his_layer()); auto status = check_seq_no(in_seq_no, out_seq_no, message->his_layer());
if (status.is_error() && status.code() != 2 /* not gap found */) { if (status.is_error() && status.code() != 2 /* not gap found */) {
message->promise.set_value(Unit()); message->promise.set_value(Unit());
if (message->logevent_id()) { if (message->log_event_id()) {
LOG(INFO) << "Erase binlog event: " << tag("logevent_id", message->logevent_id()); LOG(INFO) << "Erase binlog event: " << tag("log_event_id", message->log_event_id());
binlog_erase(context_->binlog(), message->logevent_id()); binlog_erase(context_->binlog(), message->log_event_id());
} }
auto warning_message = PSTRING() << status << tag("seq_no_state_.my_in_seq_no", seq_no_state_.my_in_seq_no) auto warning_message = PSTRING() << status << tag("seq_no_state_.my_in_seq_no", seq_no_state_.my_in_seq_no)
<< tag("seq_no_state_.my_out_seq_no", seq_no_state_.my_out_seq_no) << tag("seq_no_state_.my_out_seq_no", seq_no_state_.my_out_seq_no)
@ -1049,7 +1049,7 @@ Status SecretChatActor::do_inbound_message_decrypted_unchecked(unique_ptr<logeve
return do_inbound_message_decrypted(std::move(message)); return do_inbound_message_decrypted(std::move(message));
} }
void SecretChatActor::do_outbound_message_impl(unique_ptr<logevent::OutboundSecretMessage> binlog_event, void SecretChatActor::do_outbound_message_impl(unique_ptr<log_event::OutboundSecretMessage> binlog_event,
Promise<> promise) { Promise<> promise) {
binlog_event->crc = crc64(binlog_event->encrypted_message.as_slice()); binlog_event->crc = crc64(binlog_event->encrypted_message.as_slice());
LOG(INFO) << "Do outbound message: " << *binlog_event << tag("crc", binlog_event->crc); LOG(INFO) << "Do outbound message: " << *binlog_event << tag("crc", binlog_event->crc);
@ -1064,13 +1064,13 @@ void SecretChatActor::do_outbound_message_impl(unique_ptr<logevent::OutboundSecr
// OutboundSecretMessage // OutboundSecretMessage
// //
// 1. [] => Save logevent. [save_logevent] // 1. [] => Save log_event. [save_log_event]
// 2. [save_logevent] => Save SeqNoState [save_changes] // 2. [save_log_event] => Save SeqNoState [save_changes]
// 3. [save_logevent] => Send NetQuery [send_message] // 3. [save_log_event] => Send NetQuery [send_message]
// Note: we have to force binlog to flush // Note: we have to force binlog to flush
// 4.0 [send_message]:Fail => rewrite // 4.0 [send_message]:Fail => rewrite
// 4. [save_changes; send_message] => Mark logevent as sent [rewrite_logevent] // 4. [save_changes; send_message] => Mark log event as sent [rewrite_log_event]
// 5. [save_changes; send_message; ack] => [remove_logevent] // 5. [save_changes; send_message; ack] => [remove_log_event]
auto message = state->message.get(); auto message = state->message.get();
@ -1112,20 +1112,20 @@ void SecretChatActor::do_outbound_message_impl(unique_ptr<logevent::OutboundSecr
out_seq_no_to_outbound_message_state_token_[out_seq_no] = state_id; out_seq_no_to_outbound_message_state_token_[out_seq_no] = state_id;
} }
// save_logevent => [send_message; save_changes] // save_log_event => [send_message; save_changes]
auto save_logevent_finish = PromiseCreator::join(std::move(send_message_start), std::move(save_changes_start)); auto save_log_event_finish = PromiseCreator::join(std::move(send_message_start), std::move(save_changes_start));
auto logevent_id = state->message->logevent_id(); auto log_event_id = state->message->log_event_id();
if (logevent_id == 0) { if (log_event_id == 0) {
logevent_id = binlog_add(context_->binlog(), LogEvent::HandlerType::SecretChats, create_storer(*state->message)); log_event_id = binlog_add(context_->binlog(), LogEvent::HandlerType::SecretChats, create_storer(*state->message));
LOG(INFO) << "Outbound secret message [save_logevent] start " << tag("logevent_id", logevent_id); LOG(INFO) << "Outbound secret message [save_log_event] start " << tag("log_event_id", log_event_id);
context_->binlog()->force_sync(std::move(save_logevent_finish)); context_->binlog()->force_sync(std::move(save_log_event_finish));
state->message->set_logevent_id(logevent_id); state->message->set_log_event_id(log_event_id);
} else { } else {
LOG(INFO) << "Outbound secret message [save_logevent] skip " << tag("logevent_id", logevent_id); LOG(INFO) << "Outbound secret message [save_log_event] skip " << tag("log_event_id", log_event_id);
save_logevent_finish.set_value(Unit()); save_log_event_finish.set_value(Unit());
} }
promise.set_value(Unit()); // logevent was sent to binlog; promise.set_value(Unit()); // log event was sent to binlog
} }
void SecretChatActor::on_his_in_seq_no_updated() { void SecretChatActor::on_his_in_seq_no_updated() {
@ -1197,66 +1197,66 @@ void SecretChatActor::update_seq_no_state(const T &new_seq_no_state) {
return on_seq_no_state_changed(); return on_seq_no_state_changed();
} }
void SecretChatActor::do_inbound_message_decrypted_pending(unique_ptr<logevent::InboundSecretMessage> message) { void SecretChatActor::do_inbound_message_decrypted_pending(unique_ptr<log_event::InboundSecretMessage> message) {
// Just save logevent if necessary // Just save log event if necessary
auto logevent_id = message->logevent_id(); auto log_event_id = message->log_event_id();
// qts // qts
auto qts_promise = std::move(message->promise); auto qts_promise = std::move(message->promise);
if (logevent_id == 0) { if (log_event_id == 0) {
message->is_pending = true; message->is_pending = true;
message->set_logevent_id(binlog_add(context_->binlog(), LogEvent::HandlerType::SecretChats, create_storer(*message), message->set_log_event_id(binlog_add(context_->binlog(), LogEvent::HandlerType::SecretChats,
std::move(qts_promise))); create_storer(*message), std::move(qts_promise)));
LOG(INFO) << "Inbound PENDING secret message [save_logevent] start (do not expect finish) " LOG(INFO) << "Inbound PENDING secret message [save_log_event] start (do not expect finish) "
<< tag("logevent_id", message->logevent_id()); << tag("log_event_id", message->log_event_id());
} else { } else {
LOG(INFO) << "Inbound PENDING secret message [save_logevent] skip " << tag("logevent_id", logevent_id); LOG(INFO) << "Inbound PENDING secret message [save_log_event] skip " << tag("log_event_id", log_event_id);
CHECK(!qts_promise); CHECK(!qts_promise);
} }
LOG(INFO) << "Inbound PENDING secret message start " << tag("logevent_id", logevent_id) << tag("message", *message); LOG(INFO) << "Inbound PENDING secret message start " << tag("log_event_id", log_event_id) << tag("message", *message);
auto seq_no = message->decrypted_message_layer->out_seq_no_ / 2; auto seq_no = message->decrypted_message_layer->out_seq_no_ / 2;
pending_inbound_messages_[seq_no] = std::move(message); pending_inbound_messages_[seq_no] = std::move(message);
} }
Status SecretChatActor::do_inbound_message_decrypted(unique_ptr<logevent::InboundSecretMessage> message) { Status SecretChatActor::do_inbound_message_decrypted(unique_ptr<log_event::InboundSecretMessage> message) {
// InboundSecretMessage // InboundSecretMessage
// //
// 1. [] => Add logevent. [save_logevent] // 1. [] => Add log event. [save_log_event]
// 2. [save_logevent] => Save SeqNoState [save_changes] // 2. [save_log_event] => Save SeqNoState [save_changes]
// 3. [save_logevent] => Add message to MessageManager [save_message] // 3. [save_log_event] => Add message to MessageManager [save_message]
// Note: if we are able to add message by random_id, we may not wait for (logevent). Otherwise we should force // Note: if we are able to add message by random_id, we may not wait for (log event). Otherwise we should force
// binlog flush. // binlog flush.
// 4. [save_logevent] => Update qts [qts] // 4. [save_log_event] => Update qts [qts]
// 5. [save_changes; save_message; ?qts) => Remove logevent [remove_logevent] // 5. [save_changes; save_message; ?qts) => Remove log event [remove_log_event]
// Note: It is easier not to wait for qts. In the worst case old update will be handled again after restart. // Note: It is easier not to wait for qts. In the worst case old update will be handled again after restart.
auto state_id = inbound_message_states_.create(); auto state_id = inbound_message_states_.create();
InboundMessageState &state = *inbound_message_states_.get(state_id); InboundMessageState &state = *inbound_message_states_.get(state_id);
// save logevent // save log event
auto logevent_id = message->logevent_id(); auto log_event_id = message->log_event_id();
bool need_sync = false; bool need_sync = false;
if (logevent_id == 0) { if (log_event_id == 0) {
logevent_id = binlog_add(context_->binlog(), LogEvent::HandlerType::SecretChats, create_storer(*message)); log_event_id = binlog_add(context_->binlog(), LogEvent::HandlerType::SecretChats, create_storer(*message));
LOG(INFO) << "Inbound secret message [save_logevent] start " << tag("logevent_id", logevent_id); LOG(INFO) << "Inbound secret message [save_log_event] start " << tag("log_event_id", log_event_id);
need_sync = true; need_sync = true;
} else { } else {
if (message->is_pending) { if (message->is_pending) {
message->is_pending = false; message->is_pending = false;
auto old_logevent_id = logevent_id; auto old_log_event_id = log_event_id;
logevent_id = binlog_add(context_->binlog(), LogEvent::HandlerType::SecretChats, create_storer(*message)); log_event_id = binlog_add(context_->binlog(), LogEvent::HandlerType::SecretChats, create_storer(*message));
binlog_erase(context_->binlog(), old_logevent_id); binlog_erase(context_->binlog(), old_log_event_id);
LOG(INFO) << "Inbound secret message [save_logevent] rewrite (after pending state) " LOG(INFO) << "Inbound secret message [save_log_event] rewrite (after pending state) "
<< tag("logevent_id", logevent_id) << tag("old_logevent_id", old_logevent_id); << tag("log_event_id", log_event_id) << tag("old_log_event_id", old_log_event_id);
need_sync = true; need_sync = true;
} else { } else {
LOG(INFO) << "Inbound secret message [save_logevent] skip " << tag("logevent_id", logevent_id); LOG(INFO) << "Inbound secret message [save_log_event] skip " << tag("log_event_id", log_event_id);
} }
} }
LOG(INFO) << "Inbound secret message start " << tag("logevent_id", logevent_id) << tag("message", *message); LOG(INFO) << "Inbound secret message start " << tag("log_event_id", log_event_id) << tag("message", *message);
state.logevent_id = logevent_id; state.log_event_id = log_event_id;
// save_message // save_message
auto save_message_finish = PromiseCreator::lambda([actor_id = actor_id(this), state_id](Result<> result) { auto save_message_finish = PromiseCreator::lambda([actor_id = actor_id(this), state_id](Result<> result) {
@ -1305,7 +1305,7 @@ Status SecretChatActor::do_inbound_message_decrypted(unique_ptr<logevent::Inboun
// Send update through context_ // Send update through context_
// Note, that update may be sent multiple times and should be somehow protected from replay. // Note, that update may be sent multiple times and should be somehow protected from replay.
// Luckily all updates seems to be idempotent. // Luckily all updates seems to be idempotent.
// We could use ChangesProcessor to mark logevent as sent to context_, but I don't see any advantages of this // We could use ChangesProcessor to mark log event as sent to context_, but I don't see any advantages of this
// approach. // approach.
if (message->decrypted_message_layer->message_->get_id() == secret_api::decryptedMessage::ID) { if (message->decrypted_message_layer->message_->get_id() == secret_api::decryptedMessage::ID) {
auto decrypted_message = auto decrypted_message =
@ -1319,7 +1319,7 @@ Status SecretChatActor::do_inbound_message_decrypted(unique_ptr<logevent::Inboun
auto action = std::move(decrypted_message_service->action_); auto action = std::move(decrypted_message_service->action_);
switch (action->get_id()) { switch (action->get_id()) {
case secret_api::decryptedMessageActionDeleteMessages::ID: case secret_api::decryptedMessageActionDeleteMessages::ID:
// Corresponding logevent won't be deleted before promise returned by add_changes is set. // Corresponding log event won't be deleted before promise returned by add_changes is set.
context_->on_delete_messages( context_->on_delete_messages(
static_cast<const secret_api::decryptedMessageActionDeleteMessages &>(*action).random_ids_, static_cast<const secret_api::decryptedMessageActionDeleteMessages &>(*action).random_ids_,
std::move(save_message_finish)); std::move(save_message_finish));
@ -1385,13 +1385,13 @@ decryptedMessageActionNoop#a82fdd63 = DecryptedMessageAction;
}); });
auto save_changes_start = add_changes(std::move(save_changes_finish)); auto save_changes_start = add_changes(std::move(save_changes_finish));
// save_logevent // save_log_event
auto save_logevent_finish = PromiseCreator::join(std::move(save_changes_start), std::move(qts_promise)); auto save_log_event_finish = PromiseCreator::join(std::move(save_changes_start), std::move(qts_promise));
if (need_sync) { if (need_sync) {
// TODO: lazy sync is enough // TODO: lazy sync is enough
context_->binlog()->force_sync(std::move(save_logevent_finish)); context_->binlog()->force_sync(std::move(save_log_event_finish));
} else { } else {
save_logevent_finish.set_value(Unit()); save_log_event_finish.set_value(Unit());
} }
return Status::OK(); return Status::OK();
} }
@ -1433,7 +1433,7 @@ void SecretChatActor::on_inbound_save_message_finish(uint64 state_id) {
} }
auto *state = inbound_message_states_.get(state_id); auto *state = inbound_message_states_.get(state_id);
CHECK(state); CHECK(state);
LOG(INFO) << "Inbound message [save_message] finish " << tag("logevent_id", state->logevent_id); LOG(INFO) << "Inbound message [save_message] finish " << tag("log_event_id", state->log_event_id);
state->save_message_finish = true; state->save_message_finish = true;
inbound_loop(state, state_id); inbound_loop(state, state_id);
} }
@ -1444,7 +1444,7 @@ void SecretChatActor::on_inbound_save_changes_finish(uint64 state_id) {
} }
auto *state = inbound_message_states_.get(state_id); auto *state = inbound_message_states_.get(state_id);
CHECK(state); CHECK(state);
LOG(INFO) << "Inbound message [save_changes] finish " << tag("logevent_id", state->logevent_id); LOG(INFO) << "Inbound message [save_changes] finish " << tag("log_event_id", state->log_event_id);
state->save_changes_finish = true; state->save_changes_finish = true;
inbound_loop(state, state_id); inbound_loop(state, state_id);
} }
@ -1456,13 +1456,13 @@ void SecretChatActor::inbound_loop(InboundMessageState *state, uint64 state_id)
if (!state->save_changes_finish || !state->save_message_finish) { if (!state->save_changes_finish || !state->save_message_finish) {
return; return;
} }
LOG(INFO) << "Inbound message [remove_logevent] start " << tag("logevent_id", state->logevent_id); LOG(INFO) << "Inbound message [remove_log_event] start " << tag("log_event_id", state->log_event_id);
binlog_erase(context_->binlog(), state->logevent_id); binlog_erase(context_->binlog(), state->log_event_id);
inbound_message_states_.erase(state_id); inbound_message_states_.erase(state_id);
} }
NetQueryPtr SecretChatActor::create_net_query(const logevent::OutboundSecretMessage &message) { NetQueryPtr SecretChatActor::create_net_query(const log_event::OutboundSecretMessage &message) {
NetQueryPtr query; NetQueryPtr query;
if (message.need_notify_user) { if (message.need_notify_user) {
CHECK(message.file.empty()); CHECK(message.file.empty());
@ -1510,14 +1510,14 @@ void SecretChatActor::on_outbound_send_message_start(uint64 state_id) {
auto *message = state->message.get(); auto *message = state->message.get();
if (!message->is_sent) { if (!message->is_sent) {
LOG(INFO) << "Outbound message [send_message] start " << tag("logevent_id", state->message->logevent_id()); LOG(INFO) << "Outbound message [send_message] start " << tag("log_event_id", state->message->log_event_id());
auto query = create_net_query(*message); auto query = create_net_query(*message);
state->net_query_id = query->id(); state->net_query_id = query->id();
state->net_query_ref = query.get_weak(); state->net_query_ref = query.get_weak();
state->net_query_may_fail = state->message->is_rewritable; state->net_query_may_fail = state->message->is_rewritable;
context_->send_net_query(std::move(query), actor_shared(this, state_id), true); context_->send_net_query(std::move(query), actor_shared(this, state_id), true);
} else { } else {
LOG(INFO) << "Outbound message [send_message] start dummy " << tag("logevent_id", state->message->logevent_id()); LOG(INFO) << "Outbound message [send_message] start dummy " << tag("log_event_id", state->message->log_event_id());
on_outbound_send_message_finish(state_id); on_outbound_send_message_finish(state_id);
} }
} }
@ -1532,10 +1532,10 @@ void SecretChatActor::outbound_resend(uint64 state_id) {
state->message->is_sent = false; state->message->is_sent = false;
state->net_query_id = 0; state->net_query_id = 0;
state->net_query_ref = NetQueryRef(); state->net_query_ref = NetQueryRef();
LOG(INFO) << "Outbound message [resend] " << tag("logevent_id", state->message->logevent_id()) LOG(INFO) << "Outbound message [resend] " << tag("log_event_id", state->message->log_event_id())
<< tag("state_id", state_id); << tag("state_id", state_id);
binlog_rewrite(context_->binlog(), state->message->logevent_id(), LogEvent::HandlerType::SecretChats, binlog_rewrite(context_->binlog(), state->message->log_event_id(), LogEvent::HandlerType::SecretChats,
create_storer(*state->message)); create_storer(*state->message));
auto send_message_start = PromiseCreator::lambda([actor_id = actor_id(this), state_id](Result<> result) { auto send_message_start = PromiseCreator::lambda([actor_id = actor_id(this), state_id](Result<> result) {
if (result.is_ok()) { if (result.is_ok()) {
@ -1574,8 +1574,8 @@ Status SecretChatActor::outbound_rewrite_with_empty(uint64 state_id) {
state->message->is_external = false; state->message->is_external = false;
state->message->need_notify_user = false; state->message->need_notify_user = false;
state->message->is_silent = true; state->message->is_silent = true;
state->message->file = logevent::EncryptedInputFile::from_input_encrypted_file(nullptr); state->message->file = log_event::EncryptedInputFile::from_input_encrypted_file(nullptr);
binlog_rewrite(context_->binlog(), state->message->logevent_id(), LogEvent::HandlerType::SecretChats, binlog_rewrite(context_->binlog(), state->message->log_event_id(), LogEvent::HandlerType::SecretChats,
create_storer(*state->message)); create_storer(*state->message));
return Status::OK(); return Status::OK();
} }
@ -1593,7 +1593,7 @@ void SecretChatActor::on_outbound_send_message_result(NetQueryPtr query, Promise
} }
CHECK(state); CHECK(state);
if (state->net_query_id != query->id()) { if (state->net_query_id != query->id()) {
LOG(INFO) << "Ignore old net query result " << tag("logevent_id", state->message->logevent_id()) LOG(INFO) << "Ignore old net query result " << tag("log_event_id", state->message->log_event_id())
<< tag("query_id", query->id()) << tag("state_query_id", state->net_query_id) << query; << tag("query_id", query->id()) << tag("state_query_id", state->net_query_id) << query;
query->clear(); query->clear();
return; return;
@ -1620,7 +1620,7 @@ void SecretChatActor::on_outbound_send_message_result(NetQueryPtr query, Promise
if (state->message->is_external) { if (state->message->is_external) {
LOG(INFO) << "Outbound secret message [send_message] failed, rewrite it with dummy " LOG(INFO) << "Outbound secret message [send_message] failed, rewrite it with dummy "
<< tag("logevent_id", state->message->logevent_id()) << tag("error", error); << tag("log_event_id", state->message->log_event_id()) << tag("error", error);
state->send_result_ = [this, random_id = state->message->random_id, error_code = error.code(), state->send_result_ = [this, random_id = state->message->random_id, error_code = error.code(),
error_message = error.message()](Promise<> promise) { error_message = error.message()](Promise<> promise) {
this->context_->on_send_message_error(random_id, Status::Error(error_code, error_message), std::move(promise)); this->context_->on_send_message_error(random_id, Status::Error(error_code, error_message), std::move(promise));
@ -1629,7 +1629,7 @@ void SecretChatActor::on_outbound_send_message_result(NetQueryPtr query, Promise
} else { } else {
// Just resend. // Just resend.
LOG(INFO) << "Outbound secret message [send_message] failed, resend it " LOG(INFO) << "Outbound secret message [send_message] failed, resend it "
<< tag("logevent_id", state->message->logevent_id()) << tag("error", error); << tag("log_event_id", state->message->log_event_id()) << tag("error", error);
send_message_error_promise.set_value(Unit()); send_message_error_promise.set_value(Unit());
} }
return; return;
@ -1666,14 +1666,14 @@ void SecretChatActor::on_outbound_send_message_result(NetQueryPtr query, Promise
telegram_api::downcast_call( telegram_api::downcast_call(
*sent->file_, overloaded( *sent->file_, overloaded(
[&](telegram_api::encryptedFileEmpty &) { [&](telegram_api::encryptedFileEmpty &) {
state->message->file = logevent::EncryptedInputFile::from_input_encrypted_file( state->message->file = log_event::EncryptedInputFile::from_input_encrypted_file(
telegram_api::inputEncryptedFileEmpty()); telegram_api::inputEncryptedFileEmpty());
get_file = [] { get_file = [] {
return telegram_api::make_object<telegram_api::encryptedFileEmpty>(); return telegram_api::make_object<telegram_api::encryptedFileEmpty>();
}; };
}, },
[&](telegram_api::encryptedFile &file) { [&](telegram_api::encryptedFile &file) {
state->message->file = logevent::EncryptedInputFile::from_input_encrypted_file( state->message->file = log_event::EncryptedInputFile::from_input_encrypted_file(
telegram_api::inputEncryptedFile(file.id_, file.access_hash_)); telegram_api::inputEncryptedFile(file.id_, file.access_hash_));
get_file = [id = file.id_, access_hash = file.access_hash_, size = file.size_, get_file = [id = file.id_, access_hash = file.access_hash_, size = file.size_,
dc_id = file.dc_id_, key_fingerprint = file.key_fingerprint_] { dc_id = file.dc_id_, key_fingerprint = file.key_fingerprint_] {
@ -1758,7 +1758,7 @@ void SecretChatActor::on_outbound_send_message_finish(uint64 state_id) {
if (!state) { if (!state) {
return; return;
} }
LOG(INFO) << "Outbound secret message [send_message] finish " << tag("logevent_id", state->message->logevent_id()); LOG(INFO) << "Outbound secret message [send_message] finish " << tag("log_event_id", state->message->log_event_id());
state->send_message_finish_flag = true; state->send_message_finish_flag = true;
state->outer_send_message_finish.set_value(Unit()); state->outer_send_message_finish.set_value(Unit());
@ -1771,7 +1771,7 @@ void SecretChatActor::on_outbound_save_changes_finish(uint64 state_id) {
} }
auto *state = outbound_message_states_.get(state_id); auto *state = outbound_message_states_.get(state_id);
CHECK(state); CHECK(state);
LOG(INFO) << "Outbound secret message [save_changes] finish " << tag("logevent_id", state->message->logevent_id()); LOG(INFO) << "Outbound secret message [save_changes] finish " << tag("log_event_id", state->message->log_event_id());
state->save_changes_finish_flag = true; state->save_changes_finish_flag = true;
outbound_loop(state, state_id); outbound_loop(state, state_id);
} }
@ -1782,7 +1782,7 @@ void SecretChatActor::on_outbound_ack(uint64 state_id) {
} }
auto *state = outbound_message_states_.get(state_id); auto *state = outbound_message_states_.get(state_id);
CHECK(state); CHECK(state);
LOG(INFO) << "Outbound secret message [ack] finish " << tag("logevent_id", state->message->logevent_id()); LOG(INFO) << "Outbound secret message [ack] finish " << tag("log_event_id", state->message->log_event_id());
state->ack_flag = true; state->ack_flag = true;
outbound_loop(state, state_id); outbound_loop(state, state_id);
} }
@ -1794,7 +1794,7 @@ void SecretChatActor::on_outbound_outer_send_message_promise(uint64 state_id, Pr
} }
auto *state = outbound_message_states_.get(state_id); auto *state = outbound_message_states_.get(state_id);
CHECK(state); CHECK(state);
LOG(INFO) << "Outbound secret message [TODO] " << tag("logevent_id", state->message->logevent_id()); LOG(INFO) << "Outbound secret message [TODO] " << tag("log_event_id", state->message->log_event_id());
promise.set_value(Unit()); // Seems like this message is at least stored to binlog already promise.set_value(Unit()); // Seems like this message is at least stored to binlog already
if (state->send_result_) { if (state->send_result_) {
state->send_result_({}); state->send_result_({});
@ -1809,20 +1809,20 @@ void SecretChatActor::outbound_loop(OutboundMessageState *state, uint64 state_id
return; return;
} }
if (state->save_changes_finish_flag /*&& state->send_message_finish_flag*/ && state->ack_flag) { if (state->save_changes_finish_flag /*&& state->send_message_finish_flag*/ && state->ack_flag) {
LOG(INFO) << "Outbound message [remove_logevent] start " << tag("logevent_id", state->message->logevent_id()); LOG(INFO) << "Outbound message [remove_log_event] start " << tag("log_event_id", state->message->log_event_id());
binlog_erase(context_->binlog(), state->message->logevent_id()); binlog_erase(context_->binlog(), state->message->log_event_id());
random_id_to_outbound_message_state_token_.erase(state->message->random_id); random_id_to_outbound_message_state_token_.erase(state->message->random_id);
LOG(INFO) << "Outbound message finish (lazy) " << tag("logevent_id", state->message->logevent_id()); LOG(INFO) << "Outbound message finish (lazy) " << tag("log_event_id", state->message->log_event_id());
outbound_message_states_.erase(state_id); outbound_message_states_.erase(state_id);
return; return;
} }
if (state->save_changes_finish_flag && state->send_message_finish_flag && if (state->save_changes_finish_flag && state->send_message_finish_flag &&
!state->message->is_sent) { // [rewrite_logevent] !state->message->is_sent) { // [rewrite_log_event]
LOG(INFO) << "Outbound message [rewrite_logevent] start " << tag("logevent_id", state->message->logevent_id()); LOG(INFO) << "Outbound message [rewrite_log_event] start " << tag("log_event_id", state->message->log_event_id());
state->message->is_sent = true; state->message->is_sent = true;
binlog_rewrite(context_->binlog(), state->message->logevent_id(), LogEvent::HandlerType::SecretChats, binlog_rewrite(context_->binlog(), state->message->log_event_id(), LogEvent::HandlerType::SecretChats,
create_storer(*state->message)); create_storer(*state->message));
} }
} }
@ -1884,9 +1884,9 @@ Status SecretChatActor::on_update_chat(telegram_api::encryptedChat &update) {
return Status::Error("Key fingerprint mismatch"); return Status::Error("Key fingerprint mismatch");
} }
auth_state_.state = State::Ready; auth_state_.state = State::Ready;
if (create_logevent_id_ != 0) { if (create_log_event_id_ != 0) {
binlog_erase(context_->binlog(), create_logevent_id_); binlog_erase(context_->binlog(), create_log_event_id_);
create_logevent_id_ = 0; create_log_event_id_ = 0;
} }
// NB: order is important // NB: order is important
@ -2044,7 +2044,7 @@ void SecretChatActor::on_outbound_action(secret_api::decryptedMessageActionReadM
// TODO // TODO
} }
void SecretChatActor::on_outbound_action(secret_api::decryptedMessageActionDeleteMessages &delete_messages) { void SecretChatActor::on_outbound_action(secret_api::decryptedMessageActionDeleteMessages &delete_messages) {
// Corresponding logevent won't be deleted before promise returned by add_changes is set. // Corresponding log event won't be deleted before promise returned by add_changes is set.
on_delete_messages(delete_messages.random_ids_).ensure(); on_delete_messages(delete_messages.random_ids_).ensure();
} }
void SecretChatActor::on_outbound_action(secret_api::decryptedMessageActionScreenshotMessages &screenshot) { void SecretChatActor::on_outbound_action(secret_api::decryptedMessageActionScreenshotMessages &screenshot) {

38
td/telegram/SecretChatActor.h
View File

@ -119,10 +119,10 @@ class SecretChatActor : public NetQueryCallback {
// Inbound messages // Inbound messages
// Logevent is created by SecretChatsManager, because it must contain qts // Logevent is created by SecretChatsManager, because it must contain qts
void add_inbound_message(unique_ptr<logevent::InboundSecretMessage> message); void add_inbound_message(unique_ptr<log_event::InboundSecretMessage> message);
// Outbound messages // Outbound messages
// Promise will be set just after corresponding logevent will be SENT to binlog. // Promise will be set just after corresponding log event will be SENT to binlog.
void send_message(tl_object_ptr<secret_api::DecryptedMessage> message, void send_message(tl_object_ptr<secret_api::DecryptedMessage> message,
tl_object_ptr<telegram_api::InputEncryptedFile> file, Promise<> promise); tl_object_ptr<telegram_api::InputEncryptedFile> file, Promise<> promise);
void send_message_action(tl_object_ptr<secret_api::SendMessageAction> action); void send_message_action(tl_object_ptr<secret_api::SendMessageAction> action);
@ -136,10 +136,10 @@ class SecretChatActor : public NetQueryCallback {
void send_set_ttl_message(int32 ttl, int64 random_id, Promise<> promise); void send_set_ttl_message(int32 ttl, int64 random_id, Promise<> promise);
// Binlog replay interface // Binlog replay interface
void replay_inbound_message(unique_ptr<logevent::InboundSecretMessage> message); void replay_inbound_message(unique_ptr<log_event::InboundSecretMessage> message);
void replay_outbound_message(unique_ptr<logevent::OutboundSecretMessage> message); void replay_outbound_message(unique_ptr<log_event::OutboundSecretMessage> message);
void replay_close_chat(unique_ptr<logevent::CloseSecretChat> event); void replay_close_chat(unique_ptr<log_event::CloseSecretChat> event);
void replay_create_chat(unique_ptr<logevent::CreateSecretChat> event); void replay_create_chat(unique_ptr<log_event::CreateSecretChat> event);
void binlog_replay_finish(); void binlog_replay_finish();
private: private:
@ -461,9 +461,9 @@ class SecretChatActor : public NetQueryCallback {
bool binlog_replay_finish_flag_ = false; bool binlog_replay_finish_flag_ = false;
bool close_flag_ = false; bool close_flag_ = false;
LogEvent::Id close_logevent_id_ = 0; LogEvent::Id close_log_event_id_ = 0;
LogEvent::Id create_logevent_id_ = 0; LogEvent::Id create_log_event_id_ = 0;
enum class QueryType : uint8 { DhConfig, EncryptedChat, Message, Ignore, DiscardEncryption, ReadHistory }; enum class QueryType : uint8 { DhConfig, EncryptedChat, Message, Ignore, DiscardEncryption, ReadHistory };
@ -560,19 +560,19 @@ class SecretChatActor : public NetQueryCallback {
struct InboundMessageState { struct InboundMessageState {
bool save_changes_finish = false; bool save_changes_finish = false;
bool save_message_finish = false; bool save_message_finish = false;
LogEvent::Id logevent_id = 0; LogEvent::Id log_event_id = 0;
int32 message_id; int32 message_id;
}; };
Container<InboundMessageState> inbound_message_states_; Container<InboundMessageState> inbound_message_states_;
std::map<int32, unique_ptr<logevent::InboundSecretMessage>> pending_inbound_messages_; std::map<int32, unique_ptr<log_event::InboundSecretMessage>> pending_inbound_messages_;
Result<std::tuple<uint64, BufferSlice, int32>> decrypt(BufferSlice &encrypted_message); Result<std::tuple<uint64, BufferSlice, int32>> decrypt(BufferSlice &encrypted_message);
Status do_inbound_message_encrypted(unique_ptr<logevent::InboundSecretMessage> message); Status do_inbound_message_encrypted(unique_ptr<log_event::InboundSecretMessage> message);
Status do_inbound_message_decrypted_unchecked(unique_ptr<logevent::InboundSecretMessage> message); Status do_inbound_message_decrypted_unchecked(unique_ptr<log_event::InboundSecretMessage> message);
Status do_inbound_message_decrypted(unique_ptr<logevent::InboundSecretMessage> message); Status do_inbound_message_decrypted(unique_ptr<log_event::InboundSecretMessage> message);
void do_inbound_message_decrypted_pending(unique_ptr<logevent::InboundSecretMessage> message); void do_inbound_message_decrypted_pending(unique_ptr<log_event::InboundSecretMessage> message);
void on_inbound_save_message_finish(uint64 state_id); void on_inbound_save_message_finish(uint64 state_id);
void on_inbound_save_changes_finish(uint64 state_id); void on_inbound_save_changes_finish(uint64 state_id);
@ -580,7 +580,7 @@ class SecretChatActor : public NetQueryCallback {
// OutboundMessage // OutboundMessage
struct OutboundMessageState { struct OutboundMessageState {
unique_ptr<logevent::OutboundSecretMessage> message; unique_ptr<log_event::OutboundSecretMessage> message;
Promise<> outer_send_message_finish; Promise<> outer_send_message_finish;
Promise<> send_message_finish; Promise<> send_message_finish;
@ -618,11 +618,11 @@ class SecretChatActor : public NetQueryCallback {
void send_message_impl(tl_object_ptr<secret_api::DecryptedMessage> message, void send_message_impl(tl_object_ptr<secret_api::DecryptedMessage> message,
tl_object_ptr<telegram_api::InputEncryptedFile> file, int32 flags, Promise<> promise); tl_object_ptr<telegram_api::InputEncryptedFile> file, int32 flags, Promise<> promise);
void do_outbound_message_impl(unique_ptr<logevent::OutboundSecretMessage>, Promise<> promise); void do_outbound_message_impl(unique_ptr<log_event::OutboundSecretMessage>, Promise<> promise);
Result<BufferSlice> create_encrypted_message(int32 layer, int32 my_in_seq_no, int32 my_out_seq_no, Result<BufferSlice> create_encrypted_message(int32 layer, int32 my_in_seq_no, int32 my_out_seq_no,
tl_object_ptr<secret_api::DecryptedMessage> &message); tl_object_ptr<secret_api::DecryptedMessage> &message);
NetQueryPtr create_net_query(const logevent::OutboundSecretMessage &message); NetQueryPtr create_net_query(const log_event::OutboundSecretMessage &message);
void outbound_resend(uint64 state_id); void outbound_resend(uint64 state_id);
Status outbound_rewrite_with_empty(uint64 state_id); Status outbound_rewrite_with_empty(uint64 state_id);
@ -637,7 +637,7 @@ class SecretChatActor : public NetQueryCallback {
// DiscardEncryption // DiscardEncryption
void on_fatal_error(Status status); void on_fatal_error(Status status);
void do_close_chat_impl(unique_ptr<logevent::CloseSecretChat> event); void do_close_chat_impl(unique_ptr<log_event::CloseSecretChat> event);
void on_discard_encryption_result(NetQueryPtr result); void on_discard_encryption_result(NetQueryPtr result);
// Other // Other
@ -693,7 +693,7 @@ class SecretChatActor : public NetQueryCallback {
void on_dh_config(telegram_api::messages_dhConfigNotModified &dh_not_modified); void on_dh_config(telegram_api::messages_dhConfigNotModified &dh_not_modified);
void on_dh_config(telegram_api::messages_dhConfig &dh); void on_dh_config(telegram_api::messages_dhConfig &dh);
void do_create_chat_impl(unique_ptr<logevent::CreateSecretChat> event); void do_create_chat_impl(unique_ptr<log_event::CreateSecretChat> event);
SecretChatId get_secret_chat_id() { SecretChatId get_secret_chat_id() {
return SecretChatId(auth_state_.id); return SecretChatId(auth_state_.id);

38
td/telegram/SecretChatsManager.cpp
View File

@ -188,7 +188,7 @@ void SecretChatsManager::on_new_message(tl_object_ptr<telegram_api::EncryptedMes
} }
CHECK(message_ptr != nullptr); CHECK(message_ptr != nullptr);
auto event = make_unique<logevent::InboundSecretMessage>(); auto event = make_unique<log_event::InboundSecretMessage>();
event->promise = std::move(promise); event->promise = std::move(promise);
downcast_call(*message_ptr, [&](auto &x) { downcast_call(*message_ptr, [&](auto &x) {
event->chat_id = x.chat_id_; event->chat_id = x.chat_id_;
@ -217,26 +217,26 @@ void SecretChatsManager::replay_binlog_event(BinlogEvent &&binlog_event) {
binlog_erase(G()->td_db()->get_binlog(), binlog_event.id_); binlog_erase(G()->td_db()->get_binlog(), binlog_event.id_);
return; return;
} }
auto r_message = logevent::SecretChatEvent::from_buffer_slice(binlog_event.data_as_buffer_slice()); auto r_message = log_event::SecretChatEvent::from_buffer_slice(binlog_event.data_as_buffer_slice());
LOG_IF(FATAL, r_message.is_error()) << "Failed to deserialize event: " << r_message.error(); LOG_IF(FATAL, r_message.is_error()) << "Failed to deserialize event: " << r_message.error();
auto message = r_message.move_as_ok(); auto message = r_message.move_as_ok();
message->set_logevent_id(binlog_event.id_); message->set_log_event_id(binlog_event.id_);
LOG(INFO) << "Process binlog event " << *message; LOG(INFO) << "Process binlog event " << *message;
switch (message->get_type()) { switch (message->get_type()) {
case logevent::SecretChatEvent::Type::InboundSecretMessage: case log_event::SecretChatEvent::Type::InboundSecretMessage:
return replay_inbound_message( return replay_inbound_message(unique_ptr<log_event::InboundSecretMessage>(
unique_ptr<logevent::InboundSecretMessage>(static_cast<logevent::InboundSecretMessage *>(message.release()))); static_cast<log_event::InboundSecretMessage *>(message.release())));
case logevent::SecretChatEvent::Type::OutboundSecretMessage: case log_event::SecretChatEvent::Type::OutboundSecretMessage:
return replay_outbound_message(unique_ptr<logevent::OutboundSecretMessage>( return replay_outbound_message(unique_ptr<log_event::OutboundSecretMessage>(
static_cast<logevent::OutboundSecretMessage *>(message.release()))); static_cast<log_event::OutboundSecretMessage *>(message.release())));
case logevent::SecretChatEvent::Type::CloseSecretChat: case log_event::SecretChatEvent::Type::CloseSecretChat:
return replay_close_chat( return replay_close_chat(
unique_ptr<logevent::CloseSecretChat>(static_cast<logevent::CloseSecretChat *>(message.release()))); unique_ptr<log_event::CloseSecretChat>(static_cast<log_event::CloseSecretChat *>(message.release())));
case logevent::SecretChatEvent::Type::CreateSecretChat: case log_event::SecretChatEvent::Type::CreateSecretChat:
return replay_create_chat( return replay_create_chat(
unique_ptr<logevent::CreateSecretChat>(static_cast<logevent::CreateSecretChat *>(message.release()))); unique_ptr<log_event::CreateSecretChat>(static_cast<log_event::CreateSecretChat *>(message.release())));
} }
LOG(FATAL) << "Unknown logevent type " << tag("type", format::as_hex(static_cast<int32>(message->get_type()))); LOG(FATAL) << "Unknown log event type " << tag("type", format::as_hex(static_cast<int32>(message->get_type())));
} }
void SecretChatsManager::binlog_replay_finish() { void SecretChatsManager::binlog_replay_finish() {
@ -246,34 +246,34 @@ void SecretChatsManager::binlog_replay_finish() {
} }
} }
void SecretChatsManager::replay_inbound_message(unique_ptr<logevent::InboundSecretMessage> message) { void SecretChatsManager::replay_inbound_message(unique_ptr<log_event::InboundSecretMessage> message) {
LOG(INFO) << "Replay inbound secret message in chat " << message->chat_id; LOG(INFO) << "Replay inbound secret message in chat " << message->chat_id;
auto actor = get_chat_actor(message->chat_id); auto actor = get_chat_actor(message->chat_id);
send_closure_later(actor, &SecretChatActor::replay_inbound_message, std::move(message)); send_closure_later(actor, &SecretChatActor::replay_inbound_message, std::move(message));
} }
void SecretChatsManager::add_inbound_message(unique_ptr<logevent::InboundSecretMessage> message) { void SecretChatsManager::add_inbound_message(unique_ptr<log_event::InboundSecretMessage> message) {
LOG(INFO) << "Process inbound secret message in chat " << message->chat_id; LOG(INFO) << "Process inbound secret message in chat " << message->chat_id;
auto actor = get_chat_actor(message->chat_id); auto actor = get_chat_actor(message->chat_id);
send_closure(actor, &SecretChatActor::add_inbound_message, std::move(message)); send_closure(actor, &SecretChatActor::add_inbound_message, std::move(message));
} }
void SecretChatsManager::replay_close_chat(unique_ptr<logevent::CloseSecretChat> message) { void SecretChatsManager::replay_close_chat(unique_ptr<log_event::CloseSecretChat> message) {
LOG(INFO) << "Replay close secret chat " << message->chat_id; LOG(INFO) << "Replay close secret chat " << message->chat_id;
auto actor = get_chat_actor(message->chat_id); auto actor = get_chat_actor(message->chat_id);
send_closure_later(actor, &SecretChatActor::replay_close_chat, std::move(message)); send_closure_later(actor, &SecretChatActor::replay_close_chat, std::move(message));
} }
void SecretChatsManager::replay_create_chat(unique_ptr<logevent::CreateSecretChat> message) { void SecretChatsManager::replay_create_chat(unique_ptr<log_event::CreateSecretChat> message) {
LOG(INFO) << "Replay create secret chat " << message->random_id; LOG(INFO) << "Replay create secret chat " << message->random_id;
auto actor = create_chat_actor(message->random_id); auto actor = create_chat_actor(message->random_id);
send_closure_later(actor, &SecretChatActor::replay_create_chat, std::move(message)); send_closure_later(actor, &SecretChatActor::replay_create_chat, std::move(message));
} }
void SecretChatsManager::replay_outbound_message(unique_ptr<logevent::OutboundSecretMessage> message) { void SecretChatsManager::replay_outbound_message(unique_ptr<log_event::OutboundSecretMessage> message) {
LOG(INFO) << "Replay outbound secret message in chat " << message->chat_id; LOG(INFO) << "Replay outbound secret message in chat " << message->chat_id;
auto actor = get_chat_actor(message->chat_id); auto actor = get_chat_actor(message->chat_id);

10
td/telegram/SecretChatsManager.h
View File

@ -64,11 +64,11 @@ class SecretChatsManager : public Actor {
void flush_pending_chat_updates(); void flush_pending_chat_updates();
void do_update_chat(tl_object_ptr<telegram_api::updateEncryption> update); void do_update_chat(tl_object_ptr<telegram_api::updateEncryption> update);
void replay_inbound_message(unique_ptr<logevent::InboundSecretMessage> message); void replay_inbound_message(unique_ptr<log_event::InboundSecretMessage> message);
void add_inbound_message(unique_ptr<logevent::InboundSecretMessage> message); void add_inbound_message(unique_ptr<log_event::InboundSecretMessage> message);
void replay_outbound_message(unique_ptr<logevent::OutboundSecretMessage> message); void replay_outbound_message(unique_ptr<log_event::OutboundSecretMessage> message);
void replay_close_chat(unique_ptr<logevent::CloseSecretChat> message); void replay_close_chat(unique_ptr<log_event::CloseSecretChat> message);
void replay_create_chat(unique_ptr<logevent::CreateSecretChat> message); void replay_create_chat(unique_ptr<log_event::CreateSecretChat> message);
unique_ptr<SecretChatActor::Context> make_secret_chat_context(int32 id); unique_ptr<SecretChatActor::Context> make_secret_chat_context(int32 id);
ActorId<SecretChatActor> get_chat_actor(int32 id); ActorId<SecretChatActor> get_chat_actor(int32 id);

2
td/telegram/Td.cpp
View File

@ -4198,7 +4198,7 @@ Status Td::init(DbKey key) {
// for each Actor. // for each Actor.
// //
// 2. An actor must not make some decisions before all binlog events are processed. // 2. An actor must not make some decisions before all binlog events are processed.
// For example, SecretChatActor must not send RequestKey, before it receives logevent with RequestKey and understands // For example, SecretChatActor must not send RequestKey, before it receives log event with RequestKey and understands
// that RequestKey was already sent. // that RequestKey was already sent.
// //
// 3. During replay of binlog some queries may be sent to other actors. They shouldn't process such events before all // 3. During replay of binlog some queries may be sent to other actors. They shouldn't process such events before all

2
td/telegram/TdDb.cpp
View File

@ -127,7 +127,7 @@ Status init_binlog(Binlog &binlog, string path, BinlogKeyValue<Binlog> &binlog_p
config_pmc.external_init_handle(event); config_pmc.external_init_handle(event);
break; break;
default: default:
LOG(FATAL) << "Unsupported logevent type " << event.type_; LOG(FATAL) << "Unsupported log event type " << event.type_;
} }
}; };

30
td/telegram/WebPagesManager.cpp
View File

@ -230,7 +230,7 @@ class WebPagesManager::WebPage {
FileSourceId file_source_id; FileSourceId file_source_id;
mutable uint64 logevent_id = 0; mutable uint64 log_event_id = 0;
template <class StorerT> template <class StorerT>
void store(StorerT &storer) const { void store(StorerT &storer) const {
@ -425,10 +425,10 @@ WebPageId WebPagesManager::on_get_web_page(tl_object_ptr<telegram_api::WebPage>
LOG(INFO) << "Got empty " << web_page_id; LOG(INFO) << "Got empty " << web_page_id;
const WebPage *web_page_to_delete = get_web_page(web_page_id); const WebPage *web_page_to_delete = get_web_page(web_page_id);
if (web_page_to_delete != nullptr) { if (web_page_to_delete != nullptr) {
if (web_page_to_delete->logevent_id != 0) { if (web_page_to_delete->log_event_id != 0) {
LOG(INFO) << "Erase " << web_page_id << " from binlog"; LOG(INFO) << "Erase " << web_page_id << " from binlog";
binlog_erase(G()->td_db()->get_binlog(), web_page_to_delete->logevent_id); binlog_erase(G()->td_db()->get_binlog(), web_page_to_delete->log_event_id);
web_page_to_delete->logevent_id = 0; web_page_to_delete->log_event_id = 0;
} }
if (web_page_to_delete->file_source_id.is_valid()) { if (web_page_to_delete->file_source_id.is_valid()) {
td_->file_manager_->change_files_source(web_page_to_delete->file_source_id, td_->file_manager_->change_files_source(web_page_to_delete->file_source_id,
@ -551,7 +551,7 @@ void WebPagesManager::update_web_page(unique_ptr<WebPage> web_page, WebPageId we
} }
old_instant_view = std::move(page->instant_view); old_instant_view = std::move(page->instant_view);
web_page->logevent_id = page->logevent_id; web_page->log_event_id = page->log_event_id;
} else { } else {
auto it = url_to_file_source_id_.find(web_page->url); auto it = url_to_file_source_id_.find(web_page->url);
if (it != url_to_file_source_id_.end()) { if (it != url_to_file_source_id_.end()) {
@ -1504,12 +1504,12 @@ void WebPagesManager::save_web_page(const WebPage *web_page, WebPageId web_page_
CHECK(web_page != nullptr); CHECK(web_page != nullptr);
if (!from_binlog) { if (!from_binlog) {
WebPageLogEvent logevent(web_page_id, web_page); WebPageLogEvent log_event(web_page_id, web_page);
auto storer = get_log_event_storer(logevent); auto storer = get_log_event_storer(log_event);
if (web_page->logevent_id == 0) { if (web_page->log_event_id == 0) {
web_page->logevent_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::WebPages, storer); web_page->log_event_id = binlog_add(G()->td_db()->get_binlog(), LogEvent::HandlerType::WebPages, storer);
} else { } else {
binlog_rewrite(G()->td_db()->get_binlog(), web_page->logevent_id, LogEvent::HandlerType::WebPages, storer); binlog_rewrite(G()->td_db()->get_binlog(), web_page->log_event_id, LogEvent::HandlerType::WebPages, storer);
} }
} }
@ -1540,7 +1540,7 @@ void WebPagesManager::on_binlog_web_page_event(BinlogEvent &&event) {
auto web_page = std::move(log_event.web_page_out); auto web_page = std::move(log_event.web_page_out);
CHECK(web_page != nullptr); CHECK(web_page != nullptr);
web_page->logevent_id = event.id_; web_page->log_event_id = event.id_;
update_web_page(std::move(web_page), web_page_id, true, false); update_web_page(std::move(web_page), web_page_id, true, false);
} }
@ -1561,13 +1561,13 @@ void WebPagesManager::on_save_web_page_to_database(WebPageId web_page_id, bool s
if (!success) { if (!success) {
LOG(ERROR) << "Failed to save " << web_page_id << " to database"; LOG(ERROR) << "Failed to save " << web_page_id << " to database";
save_web_page(web_page, web_page_id, web_page->logevent_id != 0); save_web_page(web_page, web_page_id, web_page->log_event_id != 0);
} else { } else {
LOG(INFO) << "Successfully saved " << web_page_id << " to database"; LOG(INFO) << "Successfully saved " << web_page_id << " to database";
if (web_page->logevent_id != 0) { if (web_page->log_event_id != 0) {
LOG(INFO) << "Erase " << web_page_id << " from binlog"; LOG(INFO) << "Erase " << web_page_id << " from binlog";
binlog_erase(G()->td_db()->get_binlog(), web_page->logevent_id); binlog_erase(G()->td_db()->get_binlog(), web_page->log_event_id);
web_page->logevent_id = 0; web_page->log_event_id = 0;
} }
} }
} }

4
td/telegram/files/FileDb.cpp
View File

@ -278,7 +278,7 @@ class FileDb : public FileDbInterface {
//LOG(DEBUG) << "By id " << id.get() << " found data " << format::as_hex_dump<4>(Slice(data_str)); //LOG(DEBUG) << "By id " << id.get() << " found data " << format::as_hex_dump<4>(Slice(data_str));
//LOG(INFO) << attempt_count; //LOG(INFO) << attempt_count;
logevent::WithVersion<TlParser> parser(data_str); log_event::WithVersion<TlParser> parser(data_str);
parser.set_version(static_cast<int32>(Version::Initial)); parser.set_version(static_cast<int32>(Version::Initial));
FileData data; FileData data;
data.parse(parser, true); data.parse(parser, true);
@ -316,7 +316,7 @@ Status fix_file_remote_location_key_bug(SqliteDb &db) {
CHECK(TlParser(key).fetch_int() == OLD_KEY_MAGIC); CHECK(TlParser(key).fetch_int() == OLD_KEY_MAGIC);
auto remote_str = PSTRING() << key.substr(4, 4) << Slice("\0\0\0\0") << key.substr(8); auto remote_str = PSTRING() << key.substr(4, 4) << Slice("\0\0\0\0") << key.substr(8);
FullRemoteFileLocation remote; FullRemoteFileLocation remote;
logevent::WithVersion<TlParser> parser(remote_str); log_event::WithVersion<TlParser> parser(remote_str);
parser.set_version(static_cast<int32>(Version::Initial)); parser.set_version(static_cast<int32>(Version::Initial));
parse(remote, parser); parse(remote, parser);
parser.fetch_end(); parser.fetch_end();

2
td/telegram/files/FileManager.cpp
View File

@ -2873,7 +2873,7 @@ Result<FileId> FileManager::from_persistent_id_v23(Slice binary, FileType file_t
} }
auto decoded_binary = zero_decode(binary); auto decoded_binary = zero_decode(binary);
FullRemoteFileLocation remote_location; FullRemoteFileLocation remote_location;
logevent::WithVersion<TlParser> parser(decoded_binary); log_event::WithVersion<TlParser> parser(decoded_binary);
parser.set_version(version); parser.set_version(version);
parse(remote_location, parser); parse(remote_location, parser);
parser.fetch_end(); parser.fetch_end();

2
td/telegram/files/FileStatsWorker.cpp
View File

@ -58,7 +58,7 @@ void scan_db(CancellationToken &token, CallbackT &&callback) {
if (value.substr(0, 2) == "@@") { if (value.substr(0, 2) == "@@") {
return true; return true;
} }
logevent::WithVersion<TlParser> parser(value); log_event::WithVersion<TlParser> parser(value);
FileData data; FileData data;
data.parse(parser, false); data.parse(parser, false);
if (parser.get_status().is_error()) { if (parser.get_status().is_error()) {

28
td/telegram/logevent/LogEvent.h
View File

@ -25,7 +25,7 @@
#include <type_traits> #include <type_traits>
namespace td { namespace td {
namespace logevent { namespace log_event {
template <class ParentT> template <class ParentT>
class WithVersion : public ParentT { class WithVersion : public ParentT {
@ -107,19 +107,19 @@ class LogEvent {
using Id = uint64; using Id = uint64;
Id logevent_id() const { Id log_event_id() const {
return logevent_id_; return log_event_id_;
} }
void set_logevent_id(Id logevent_id) { void set_log_event_id(Id log_event_id) {
logevent_id_ = logevent_id; log_event_id_ = log_event_id;
} }
virtual StringBuilder &print(StringBuilder &sb) const { virtual StringBuilder &print(StringBuilder &sb) const {
return sb << "[Logevent " << tag("id", logevent_id()) << "]"; return sb << "[Logevent " << tag("id", log_event_id()) << "]";
} }
private: private:
Id logevent_id_{}; Id log_event_id_{};
}; };
inline StringBuilder &operator<<(StringBuilder &sb, const LogEvent &log_event) { inline StringBuilder &operator<<(StringBuilder &sb, const LogEvent &log_event) {
return log_event.print(sb); return log_event.print(sb);
@ -257,12 +257,12 @@ class LogEventStorerImpl : public Storer {
const T &event_; const T &event_;
}; };
} // namespace logevent } // namespace log_event
using LogEvent = logevent::LogEvent; using LogEvent = log_event::LogEvent;
using LogEventParser = logevent::LogEventParser; using LogEventParser = log_event::LogEventParser;
using LogEventStorerCalcLength = logevent::LogEventStorerCalcLength; using LogEventStorerCalcLength = log_event::LogEventStorerCalcLength;
using LogEventStorerUnsafe = logevent::LogEventStorerUnsafe; using LogEventStorerUnsafe = log_event::LogEventStorerUnsafe;
template <class T> template <class T>
Status log_event_parse(T &data, Slice slice) TD_WARN_UNUSED_RESULT; Status log_event_parse(T &data, Slice slice) TD_WARN_UNUSED_RESULT;
@ -295,8 +295,8 @@ BufferSlice log_event_store(const T &data) {
} }
template <class T> template <class T>
logevent::LogEventStorerImpl<T> get_log_event_storer(const T &event) { log_event::LogEventStorerImpl<T> get_log_event_storer(const T &event) {
return logevent::LogEventStorerImpl<T>(event); return log_event::LogEventStorerImpl<T>(event);
} }
} // namespace td } // namespace td

40
td/telegram/logevent/LogEventHelper.cpp
View File

@ -15,43 +15,41 @@
namespace td { namespace td {
void add_log_event(LogeventIdWithGeneration &logevent_id, const Storer &storer, uint32 type, Slice name) { void add_log_event(LogeventIdWithGeneration &log_event_id, const Storer &storer, uint32 type, Slice name) {
LOG(INFO) << "Save " << name << " to binlog"; LOG(INFO) << "Save " << name << " to binlog";
if (logevent_id.logevent_id == 0) { if (log_event_id.log_event_id == 0) {
logevent_id.logevent_id = log_event_id.log_event_id = binlog_add(G()->td_db()->get_binlog(), type, storer);
binlog_add(G()->td_db()->get_binlog(), type, storer); LOG(INFO) << "Add " << name << " log event " << log_event_id.log_event_id;
LOG(INFO) << "Add " << name << " logevent " << logevent_id.logevent_id;
} else { } else {
auto new_logevent_id = binlog_rewrite(G()->td_db()->get_binlog(), logevent_id.logevent_id, auto new_log_event_id = binlog_rewrite(G()->td_db()->get_binlog(), log_event_id.log_event_id, type, storer);
type, storer); LOG(INFO) << "Rewrite " << name << " log event " << log_event_id.log_event_id << " with " << new_log_event_id;
LOG(INFO) << "Rewrite " << name << " logevent " << logevent_id.logevent_id << " with " << new_logevent_id;
} }
logevent_id.generation++; log_event_id.generation++;
} }
void delete_log_event(LogeventIdWithGeneration &logevent_id, uint64 generation, Slice name) { void delete_log_event(LogeventIdWithGeneration &log_event_id, uint64 generation, Slice name) {
LOG(INFO) << "Finish to process " << name << " logevent " << logevent_id.logevent_id << " with generation " << generation; LOG(INFO) << "Finish to process " << name << " log event " << log_event_id.log_event_id << " with generation "
if (logevent_id.generation == generation) { << generation;
CHECK(logevent_id.logevent_id != 0); if (log_event_id.generation == generation) {
LOG(INFO) << "Delete " << name << " logevent " << logevent_id.logevent_id; CHECK(log_event_id.log_event_id != 0);
binlog_erase(G()->td_db()->get_binlog(), logevent_id.logevent_id); LOG(INFO) << "Delete " << name << " log event " << log_event_id.log_event_id;
logevent_id.logevent_id = 0; binlog_erase(G()->td_db()->get_binlog(), log_event_id.log_event_id);
log_event_id.log_event_id = 0;
} }
} }
Promise<Unit> get_erase_logevent_promise(uint64 logevent_id, Promise<Unit> promise) { Promise<Unit> get_erase_log_event_promise(uint64 log_event_id, Promise<Unit> promise) {
if (logevent_id == 0) { if (log_event_id == 0) {
return promise; return promise;
} }
return PromiseCreator::lambda([logevent_id, promise = std::move(promise)](Result<Unit> result) mutable { return PromiseCreator::lambda([log_event_id, promise = std::move(promise)](Result<Unit> result) mutable {
if (!G()->close_flag()) { if (!G()->close_flag()) {
binlog_erase(G()->td_db()->get_binlog(), logevent_id); binlog_erase(G()->td_db()->get_binlog(), log_event_id);
} }
promise.set_result(std::move(result)); promise.set_result(std::move(result));
}); });
} }
} // namespace td } // namespace td

8
td/telegram/logevent/LogEventHelper.h
View File

@ -16,15 +16,15 @@
namespace td { namespace td {
struct LogeventIdWithGeneration { struct LogeventIdWithGeneration {
uint64 logevent_id = 0; uint64 log_event_id = 0;
uint64 generation = 0; uint64 generation = 0;
}; };
void add_log_event(LogeventIdWithGeneration &logevent_id, const Storer &storer, uint32 type, Slice name); void add_log_event(LogeventIdWithGeneration &log_event_id, const Storer &storer, uint32 type, Slice name);
void delete_log_event(LogeventIdWithGeneration &logevent_id, uint64 generation, Slice name); void delete_log_event(LogeventIdWithGeneration &log_event_id, uint64 generation, Slice name);
Promise<Unit> get_erase_logevent_promise(uint64 logevent_id, Promise<Unit> promise = Promise<Unit>()); Promise<Unit> get_erase_log_event_promise(uint64 log_event_id, Promise<Unit> promise = Promise<Unit>());
template <class StorerT> template <class StorerT>
void store_time(double time_at, StorerT &storer) { void store_time(double time_at, StorerT &storer) {

16
td/telegram/logevent/SecretChatEvent.h
View File

@ -20,7 +20,7 @@
#include "td/telegram/telegram_api.h" #include "td/telegram/telegram_api.h"
namespace td { namespace td {
namespace logevent { namespace log_event {
class SecretChatEvent : public LogEventBase<SecretChatEvent> { class SecretChatEvent : public LogEventBase<SecretChatEvent> {
public: public:
@ -296,7 +296,7 @@ class InboundSecretMessage : public SecretChatLogEventBase<InboundSecretMessage>
} }
StringBuilder &print(StringBuilder &sb) const override { StringBuilder &print(StringBuilder &sb) const override {
return sb << "[Logevent InboundSecretMessage " << tag("id", logevent_id()) << tag("chat_id", chat_id) return sb << "[Logevent InboundSecretMessage " << tag("id", log_event_id()) << tag("chat_id", chat_id)
<< tag("date", date) << tag("auth_key_id", format::as_hex(auth_key_id)) << tag("message_id", message_id) << tag("date", date) << tag("auth_key_id", format::as_hex(auth_key_id)) << tag("message_id", message_id)
<< tag("my_in_seq_no", my_in_seq_no) << tag("my_out_seq_no", my_out_seq_no) << tag("my_in_seq_no", my_in_seq_no) << tag("my_out_seq_no", my_out_seq_no)
<< tag("his_in_seq_no", his_in_seq_no) << tag("message", to_string(decrypted_message_layer)) << tag("his_in_seq_no", his_in_seq_no) << tag("message", to_string(decrypted_message_layer))
@ -402,7 +402,7 @@ class OutboundSecretMessage : public SecretChatLogEventBase<OutboundSecretMessag
} }
StringBuilder &print(StringBuilder &sb) const override { StringBuilder &print(StringBuilder &sb) const override {
return sb << "[Logevent OutboundSecretMessage " << tag("id", logevent_id()) << tag("chat_id", chat_id) return sb << "[Logevent OutboundSecretMessage " << tag("id", log_event_id()) << tag("chat_id", chat_id)
<< tag("is_sent", is_sent) << tag("need_notify_user", need_notify_user) << tag("is_sent", is_sent) << tag("need_notify_user", need_notify_user)
<< tag("is_rewritable", is_rewritable) << tag("is_external", is_external) << tag("message_id", message_id) << tag("is_rewritable", is_rewritable) << tag("is_external", is_external) << tag("message_id", message_id)
<< tag("random_id", random_id) << tag("my_in_seq_no", my_in_seq_no) << tag("my_out_seq_no", my_out_seq_no) << tag("random_id", random_id) << tag("my_in_seq_no", my_in_seq_no) << tag("my_out_seq_no", my_out_seq_no)
@ -428,7 +428,7 @@ class CloseSecretChat : public SecretChatLogEventBase<CloseSecretChat> {
} }
StringBuilder &print(StringBuilder &sb) const override { StringBuilder &print(StringBuilder &sb) const override {
return sb << "[Logevent CloseSecretChat " << tag("id", logevent_id()) << tag("chat_id", chat_id) << "]"; return sb << "[Logevent CloseSecretChat " << tag("id", log_event_id()) << tag("chat_id", chat_id) << "]";
} }
}; };
@ -456,7 +456,7 @@ class CreateSecretChat : public SecretChatLogEventBase<CreateSecretChat> {
} }
StringBuilder &print(StringBuilder &sb) const override { StringBuilder &print(StringBuilder &sb) const override {
return sb << "[Logevent CreateSecretChat " << tag("id", logevent_id()) << tag("chat_id", random_id) return sb << "[Logevent CreateSecretChat " << tag("id", log_event_id()) << tag("chat_id", random_id)
<< tag("user_id", user_id) << "]"; << tag("user_id", user_id) << "]";
} }
}; };
@ -480,10 +480,10 @@ void SecretChatEvent::downcast_call(Type type, F &&f) {
break; break;
} }
} }
} // namespace logevent } // namespace log_event
inline auto create_storer(logevent::SecretChatEvent &event) { inline auto create_storer(log_event::SecretChatEvent &event) {
return logevent::detail::StorerImpl<logevent::SecretChatEvent>(event); return log_event::detail::StorerImpl<log_event::SecretChatEvent>(event);
} }
} // namespace td } // namespace td

36
tddb/td/db/TQueue.cpp
View File

@ -113,8 +113,8 @@ class TQueueImpl : public TQueue {
auto it = q.events.end(); auto it = q.events.end();
--it; --it;
if (it->second.data.empty()) { if (it->second.data.empty()) {
if (callback_ != nullptr && it->second.logevent_id != 0) { if (callback_ != nullptr && it->second.log_event_id != 0) {
callback_->pop(it->second.logevent_id); callback_->pop(it->second.log_event_id);
} }
q.events.erase(it); q.events.erase(it);
} }
@ -123,8 +123,8 @@ class TQueueImpl : public TQueue {
schedule_queue_gc(queue_id, q, raw_event.expires_at); schedule_queue_gc(queue_id, q, raw_event.expires_at);
} }
if (raw_event.logevent_id == 0 && callback_ != nullptr) { if (raw_event.log_event_id == 0 && callback_ != nullptr) {
raw_event.logevent_id = callback_->push(queue_id, raw_event); raw_event.log_event_id = callback_->push(queue_id, raw_event);
} }
q.tail_id = event_id.next().move_as_ok(); q.tail_id = event_id.next().move_as_ok();
q.total_event_length += raw_event.data.size(); q.total_event_length += raw_event.data.size();
@ -307,7 +307,7 @@ class TQueueImpl : public TQueue {
void pop(Queue &q, QueueId queue_id, std::map<EventId, RawEvent>::iterator &it, EventId tail_id) { void pop(Queue &q, QueueId queue_id, std::map<EventId, RawEvent>::iterator &it, EventId tail_id) {
auto &event = it->second; auto &event = it->second;
if (callback_ == nullptr || event.logevent_id == 0) { if (callback_ == nullptr || event.log_event_id == 0) {
remove_event(q, it); remove_event(q, it);
return; return;
} }
@ -319,7 +319,7 @@ class TQueueImpl : public TQueue {
} }
++it; ++it;
} else { } else {
callback_->pop(event.logevent_id); callback_->pop(event.log_event_id);
remove_event(q, it); remove_event(q, it);
} }
} }
@ -438,16 +438,16 @@ uint64 TQueueBinlog<BinlogT>::push(QueueId queue_id, const RawEvent &event) {
log_event.data = event.data; log_event.data = event.data;
log_event.extra = event.extra; log_event.extra = event.extra;
auto magic = BINLOG_EVENT_TYPE + (log_event.extra != 0); auto magic = BINLOG_EVENT_TYPE + (log_event.extra != 0);
if (event.logevent_id == 0) { if (event.log_event_id == 0) {
return binlog_->add(magic, log_event); return binlog_->add(magic, log_event);
} }
binlog_->rewrite(event.logevent_id, magic, log_event); binlog_->rewrite(event.log_event_id, magic, log_event);
return event.logevent_id; return event.log_event_id;
} }
template <class BinlogT> template <class BinlogT>
void TQueueBinlog<BinlogT>::pop(uint64 logevent_id) { void TQueueBinlog<BinlogT>::pop(uint64 log_event_id) {
binlog_->erase(logevent_id); binlog_->erase(log_event_id);
} }
template <class BinlogT> template <class BinlogT>
@ -463,7 +463,7 @@ Status TQueueBinlog<BinlogT>::replay(const BinlogEvent &binlog_event, TQueue &q)
TRY_STATUS(parser.get_status()); TRY_STATUS(parser.get_status());
TRY_RESULT(event_id, EventId::from_int32(event.event_id)); TRY_RESULT(event_id, EventId::from_int32(event.event_id));
RawEvent raw_event; RawEvent raw_event;
raw_event.logevent_id = binlog_event.id_; raw_event.log_event_id = binlog_event.id_;
raw_event.event_id = event_id; raw_event.event_id = event_id;
raw_event.expires_at = event.expires_at; raw_event.expires_at = event.expires_at;
raw_event.data = event.data.str(); raw_event.data = event.data.str();
@ -483,19 +483,19 @@ template class TQueueBinlog<BinlogInterface>;
template class TQueueBinlog<Binlog>; template class TQueueBinlog<Binlog>;
uint64 TQueueMemoryStorage::push(QueueId queue_id, const RawEvent &event) { uint64 TQueueMemoryStorage::push(QueueId queue_id, const RawEvent &event) {
auto logevent_id = event.logevent_id == 0 ? next_logevent_id_++ : event.logevent_id; auto log_event_id = event.log_event_id == 0 ? next_log_event_id_++ : event.log_event_id;
events_[logevent_id] = std::make_pair(queue_id, event); events_[log_event_id] = std::make_pair(queue_id, event);
return logevent_id; return log_event_id;
} }
void TQueueMemoryStorage::pop(uint64 logevent_id) { void TQueueMemoryStorage::pop(uint64 log_event_id) {
events_.erase(logevent_id); events_.erase(log_event_id);
} }
void TQueueMemoryStorage::replay(TQueue &q) const { void TQueueMemoryStorage::replay(TQueue &q) const {
for (auto &e : events_) { for (auto &e : events_) {
auto x = e.second; auto x = e.second;
x.second.logevent_id = e.first; x.second.log_event_id = e.first;
bool is_added = q.do_push(x.first, std::move(x.second)); bool is_added = q.do_push(x.first, std::move(x.second));
CHECK(is_added); CHECK(is_added);
} }

10
tddb/td/db/TQueue.h
View File

@ -60,7 +60,7 @@ class TQueue {
}; };
struct RawEvent { struct RawEvent {
uint64 logevent_id{0}; uint64 log_event_id{0};
EventId event_id; EventId event_id;
string data; string data;
int64 extra{0}; int64 extra{0};
@ -82,7 +82,7 @@ class TQueue {
virtual ~StorageCallback() = default; virtual ~StorageCallback() = default;
virtual uint64 push(QueueId queue_id, const RawEvent &event) = 0; virtual uint64 push(QueueId queue_id, const RawEvent &event) = 0;
virtual void pop(uint64 logevent_id) = 0; virtual void pop(uint64 log_event_id) = 0;
virtual void close(Promise<> promise) = 0; virtual void close(Promise<> promise) = 0;
}; };
@ -125,7 +125,7 @@ template <class BinlogT>
class TQueueBinlog : public TQueue::StorageCallback { class TQueueBinlog : public TQueue::StorageCallback {
public: public:
uint64 push(QueueId queue_id, const RawEvent &event) override; uint64 push(QueueId queue_id, const RawEvent &event) override;
void pop(uint64 logevent_id) override; void pop(uint64 log_event_id) override;
Status replay(const BinlogEvent &binlog_event, TQueue &q) const TD_WARN_UNUSED_RESULT; Status replay(const BinlogEvent &binlog_event, TQueue &q) const TD_WARN_UNUSED_RESULT;
void set_binlog(std::shared_ptr<BinlogT> binlog) { void set_binlog(std::shared_ptr<BinlogT> binlog) {
@ -141,12 +141,12 @@ class TQueueBinlog : public TQueue::StorageCallback {
class TQueueMemoryStorage : public TQueue::StorageCallback { class TQueueMemoryStorage : public TQueue::StorageCallback {
public: public:
uint64 push(QueueId queue_id, const RawEvent &event) override; uint64 push(QueueId queue_id, const RawEvent &event) override;
void pop(uint64 logevent_id) override; void pop(uint64 log_event_id) override;
void replay(TQueue &q) const; void replay(TQueue &q) const;
virtual void close(Promise<> promise) override; virtual void close(Promise<> promise) override;
private: private:
uint64 next_logevent_id_{1}; uint64 next_log_event_id_{1};
std::map<uint64, std::pair<QueueId, RawEvent>> events_; std::map<uint64, std::pair<QueueId, RawEvent>> events_;
}; };

14
tddb/td/db/binlog/Binlog.h
View File

@ -72,20 +72,20 @@ class Binlog {
} }
uint64 add(int32 type, const Storer &storer) { uint64 add(int32 type, const Storer &storer) {
auto logevent_id = next_id(); auto log_event_id = next_id();
add_raw_event(BinlogEvent::create_raw(logevent_id, type, 0, storer), {}); add_raw_event(BinlogEvent::create_raw(log_event_id, type, 0, storer), {});
return logevent_id; return log_event_id;
} }
uint64 rewrite(uint64 logevent_id, int32 type, const Storer &storer) { uint64 rewrite(uint64 log_event_id, int32 type, const Storer &storer) {
auto seq_no = next_id(); auto seq_no = next_id();
add_raw_event(BinlogEvent::create_raw(logevent_id, type, BinlogEvent::Flags::Rewrite, storer), {}); add_raw_event(BinlogEvent::create_raw(log_event_id, type, BinlogEvent::Flags::Rewrite, storer), {});
return seq_no; return seq_no;
} }
uint64 erase(uint64 logevent_id) { uint64 erase(uint64 log_event_id) {
auto seq_no = next_id(); auto seq_no = next_id();
add_raw_event(BinlogEvent::create_raw(logevent_id, BinlogEvent::ServiceTypes::Empty, BinlogEvent::Flags::Rewrite, add_raw_event(BinlogEvent::create_raw(log_event_id, BinlogEvent::ServiceTypes::Empty, BinlogEvent::Flags::Rewrite,
EmptyStorer()), EmptyStorer()),
{}); {});
return seq_no; return seq_no;

8
tddb/td/db/binlog/BinlogHelper.h
View File

@ -21,13 +21,13 @@ inline uint64 binlog_add(BinlogInterface *binlog_ptr, int32 type, const Storer &
return binlog_ptr->add(type, storer, std::move(promise)); return binlog_ptr->add(type, storer, std::move(promise));
} }
inline uint64 binlog_rewrite(BinlogInterface *binlog_ptr, uint64 logevent_id, int32 type, const Storer &storer, inline uint64 binlog_rewrite(BinlogInterface *binlog_ptr, uint64 log_event_id, int32 type, const Storer &storer,
Promise<> promise = Promise<>()) { Promise<> promise = Promise<>()) {
return binlog_ptr->rewrite(logevent_id, type, storer, std::move(promise)); return binlog_ptr->rewrite(log_event_id, type, storer, std::move(promise));
} }
inline uint64 binlog_erase(BinlogInterface *binlog_ptr, uint64 logevent_id, Promise<> promise = Promise<>()) { inline uint64 binlog_erase(BinlogInterface *binlog_ptr, uint64 log_event_id, Promise<> promise = Promise<>()) {
return binlog_ptr->erase(logevent_id, std::move(promise)); return binlog_ptr->erase(log_event_id, std::move(promise));
} }
} // namespace td } // namespace td

14
tddb/td/db/binlog/BinlogInterface.h
View File

@ -43,22 +43,22 @@ class BinlogInterface {
} }
uint64 add(int32 type, const Storer &storer, Promise<> promise = Promise<>()) { uint64 add(int32 type, const Storer &storer, Promise<> promise = Promise<>()) {
auto logevent_id = next_id(); auto log_event_id = next_id();
add_raw_event_impl(logevent_id, BinlogEvent::create_raw(logevent_id, type, 0, storer), std::move(promise), {}); add_raw_event_impl(log_event_id, BinlogEvent::create_raw(log_event_id, type, 0, storer), std::move(promise), {});
return logevent_id; return log_event_id;
} }
uint64 rewrite(uint64 logevent_id, int32 type, const Storer &storer, Promise<> promise = Promise<>()) { uint64 rewrite(uint64 log_event_id, int32 type, const Storer &storer, Promise<> promise = Promise<>()) {
auto seq_no = next_id(); auto seq_no = next_id();
add_raw_event_impl(seq_no, BinlogEvent::create_raw(logevent_id, type, BinlogEvent::Flags::Rewrite, storer), add_raw_event_impl(seq_no, BinlogEvent::create_raw(log_event_id, type, BinlogEvent::Flags::Rewrite, storer),
std::move(promise), {}); std::move(promise), {});
return seq_no; return seq_no;
} }
uint64 erase(uint64 logevent_id, Promise<> promise = Promise<>()) { uint64 erase(uint64 log_event_id, Promise<> promise = Promise<>()) {
auto seq_no = next_id(); auto seq_no = next_id();
add_raw_event_impl(seq_no, add_raw_event_impl(seq_no,
BinlogEvent::create_raw(logevent_id, BinlogEvent::ServiceTypes::Empty, BinlogEvent::create_raw(log_event_id, BinlogEvent::ServiceTypes::Empty,
BinlogEvent::Flags::Rewrite, EmptyStorer()), BinlogEvent::Flags::Rewrite, EmptyStorer()),
std::move(promise), {}); std::move(promise), {});
return seq_no; return seq_no;

2
tddb/td/db/binlog/detail/BinlogEventsProcessor.cpp
View File

@ -20,7 +20,7 @@ Status BinlogEventsProcessor::do_event(BinlogEvent &&event) {
if ((event.flags_ & BinlogEvent::Flags::Rewrite) && !ids_.empty() && ids_.back() >= fixed_id) { if ((event.flags_ & BinlogEvent::Flags::Rewrite) && !ids_.empty() && ids_.back() >= fixed_id) {
auto it = std::lower_bound(ids_.begin(), ids_.end(), fixed_id); auto it = std::lower_bound(ids_.begin(), ids_.end(), fixed_id);
if (it == ids_.end() || *it != fixed_id) { if (it == ids_.end() || *it != fixed_id) {
return Status::Error(PSLICE() << "Ignore rewrite logevent " << event.public_to_string()); return Status::Error(PSLICE() << "Ignore rewrite log event " << event.public_to_string());
} }
auto pos = it - ids_.begin(); auto pos = it - ids_.begin();
total_raw_events_size_ -= static_cast<int64>(events_[pos].raw_event_.size()); total_raw_events_size_ -= static_cast<int64>(events_[pos].raw_event_.size());

18
test/secret.cpp
View File

@ -589,7 +589,7 @@ class Master : public Actor {
void add_inbound_message(int32 chat_id, BufferSlice data, uint64 crc) { void add_inbound_message(int32 chat_id, BufferSlice data, uint64 crc) {
CHECK(crc64(data.as_slice()) == crc); CHECK(crc64(data.as_slice()) == crc);
auto event = make_unique<logevent::InboundSecretMessage>(); auto event = make_unique<log_event::InboundSecretMessage>();
event->chat_id = chat_id; event->chat_id = chat_id;
event->date = 0; event->date = 0;
event->encrypted_message = std::move(data); event->encrypted_message = std::move(data);
@ -668,21 +668,21 @@ class Master : public Actor {
for (auto &event : events) { for (auto &event : events) {
CHECK(event.type_ == LogEvent::HandlerType::SecretChats); CHECK(event.type_ == LogEvent::HandlerType::SecretChats);
auto r_message = logevent::SecretChatEvent::from_buffer_slice(event.data_as_buffer_slice()); auto r_message = log_event::SecretChatEvent::from_buffer_slice(event.data_as_buffer_slice());
LOG_IF(FATAL, r_message.is_error()) << "Failed to deserialize event: " << r_message.error(); LOG_IF(FATAL, r_message.is_error()) << "Failed to deserialize event: " << r_message.error();
auto message = r_message.move_as_ok(); auto message = r_message.move_as_ok();
message->set_logevent_id(event.id_); message->set_log_event_id(event.id_);
LOG(INFO) << "Process binlog event " << *message; LOG(INFO) << "Process binlog event " << *message;
switch (message->get_type()) { switch (message->get_type()) {
case logevent::SecretChatEvent::Type::InboundSecretMessage: case log_event::SecretChatEvent::Type::InboundSecretMessage:
send_closure_later(actor_, &SecretChatActor::replay_inbound_message, send_closure_later(actor_, &SecretChatActor::replay_inbound_message,
unique_ptr<logevent::InboundSecretMessage>( unique_ptr<log_event::InboundSecretMessage>(
static_cast<logevent::InboundSecretMessage *>(message.release()))); static_cast<log_event::InboundSecretMessage *>(message.release())));
break; break;
case logevent::SecretChatEvent::Type::OutboundSecretMessage: case log_event::SecretChatEvent::Type::OutboundSecretMessage:
send_closure_later(actor_, &SecretChatActor::replay_outbound_message, send_closure_later(actor_, &SecretChatActor::replay_outbound_message,
unique_ptr<logevent::OutboundSecretMessage>( unique_ptr<log_event::OutboundSecretMessage>(
static_cast<logevent::OutboundSecretMessage *>(message.release()))); static_cast<log_event::OutboundSecretMessage *>(message.release())));
break; break;
default: default:
UNREACHABLE(); UNREACHABLE();