tdlight-team/tdlight
tdlight-team/tdlight
1
1
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
46ed865570
tdlight/td/telegram/MessagesDb.cpp
levlam b13096e8a9 Fix adding to database of rescheduled messages. 
GitOrigin-RevId: fe07906bdc9ef264d9d40e225dbc8fbb7d547acd
2019-12-06 21:46:57 +03:00

1170 lines
48 KiB
C++
Raw Blame History

//
// Copyright Aliaksei Levin (levlam@telegram.org), Arseny Smirnov (arseny30@gmail.com) 2014-2019
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//
#include "td/telegram/MessagesDb.h"
#include "td/telegram/logevent/LogEvent.h"
#include "td/telegram/Version.h"
#include "td/db/SqliteConnectionSafe.h"
#include "td/db/SqliteDb.h"
#include "td/db/SqliteStatement.h"
#include "td/actor/actor.h"
#include "td/actor/PromiseFuture.h"
#include "td/actor/SchedulerLocalStorage.h"
#include "td/utils/format.h"
#include "td/utils/logging.h"
#include "td/utils/ScopeGuard.h"
#include "td/utils/Slice.h"
#include "td/utils/StackAllocator.h"
#include "td/utils/StringBuilder.h"
#include "td/utils/Time.h"
#include "td/utils/tl_helpers.h"
#include "td/utils/unicode.h"
#include "td/utils/utf8.h"
#include <algorithm>
#include <array>
#include <iterator>
#include <limits>
#include <tuple>
namespace td {
static constexpr int32 MESSAGES_DB_INDEX_COUNT = 30;
static constexpr int32 MESSAGES_DB_INDEX_COUNT_OLD = 9;
// NB: must happen inside a transaction
Status init_messages_db(SqliteDb &db, int32 version) {
LOG(INFO) << "Init message database " << tag("version", version);
// Check if database exists
TRY_RESULT(has_table, db.has_table("messages"));
if (!has_table) {
version = 0;
} else if (version < static_cast<int32>(DbVersion::DialogDbCreated) || version > current_db_version()) {
TRY_STATUS(drop_messages_db(db, version));
version = 0;
}
auto add_media_indices = [&db](int begin, int end) {
for (int i = begin; i < end; i++) {
TRY_STATUS(db.exec(PSLICE() << "CREATE INDEX IF NOT EXISTS message_index_" << i
<< " ON messages (dialog_id, message_id) WHERE (index_mask & " << (1 << i)
<< ") != 0"));
}
return Status::OK();
};
auto add_fts = [&db] {
TRY_STATUS(
db.exec("CREATE INDEX IF NOT EXISTS message_by_search_id ON messages "
"(search_id) WHERE search_id IS NOT NULL"));
TRY_STATUS(
db.exec("CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(text, content='messages', "
"content_rowid='search_id', tokenize = \"unicode61 remove_diacritics 0 tokenchars '\a'\")"));
TRY_STATUS(db.exec(
"CREATE TRIGGER IF NOT EXISTS trigger_fts_delete BEFORE DELETE ON messages WHEN OLD.search_id IS NOT NULL"
" BEGIN INSERT INTO messages_fts(messages_fts, rowid, text) VALUES(\'delete\', OLD.search_id, OLD.text); END"));
TRY_STATUS(db.exec(
"CREATE TRIGGER IF NOT EXISTS trigger_fts_insert AFTER INSERT ON messages WHEN NEW.search_id IS NOT NULL"
" BEGIN INSERT INTO messages_fts(rowid, text) VALUES(NEW.search_id, NEW.text); END"));
//TRY_STATUS(db.exec(
//"CREATE TRIGGER IF NOT EXISTS trigger_fts_update AFTER UPDATE ON messages WHEN NEW.search_id IS NOT NULL OR "
//"OLD.search_id IS NOT NULL"
//" BEGIN "
//"INSERT INTO messages_fts(messages_fts, rowid, text) VALUES(\'delete\', OLD.search_id, OLD.text); "
//"INSERT INTO messages_fts(rowid, text) VALUES(NEW.search_id, NEW.text); "
//" END"));
return Status::OK();
};
auto add_call_index = [&db]() {
for (int i = static_cast<int>(SearchMessagesFilter::Call) - 1;
i < static_cast<int>(SearchMessagesFilter::MissedCall); i++) {
TRY_STATUS(db.exec(PSLICE() << "CREATE INDEX IF NOT EXISTS full_message_index_" << i
<< " ON messages (unique_message_id) WHERE (index_mask & " << (1 << i) << ") != 0"));
}
return Status::OK();
};
auto add_notification_id_index = [&db]() {
return db.exec(
"CREATE INDEX IF NOT EXISTS message_by_notification_id ON messages (dialog_id, notification_id) WHERE "
"notification_id IS NOT NULL");
};
auto add_scheduled_messages_table = [&db]() {
TRY_STATUS(
db.exec("CREATE TABLE IF NOT EXISTS scheduled_messages (dialog_id INT8, message_id INT8, "
"server_message_id INT4, data BLOB, PRIMARY KEY (dialog_id, message_id))"));
TRY_STATUS(
db.exec("CREATE INDEX IF NOT EXISTS message_by_server_message_id ON scheduled_messages "
"(dialog_id, server_message_id) WHERE server_message_id IS NOT NULL"));
return Status::OK();
};
if (version == 0) {
LOG(INFO) << "Create new message database";
TRY_STATUS(
db.exec("CREATE TABLE IF NOT EXISTS messages (dialog_id INT8, message_id INT8, "
"unique_message_id INT4, sender_user_id INT4, random_id INT8, data BLOB, "
"ttl_expires_at INT4, index_mask INT4, search_id INT8, text STRING, notification_id INT4, PRIMARY KEY "
"(dialog_id, message_id))"));
TRY_STATUS(
db.exec("CREATE INDEX IF NOT EXISTS message_by_random_id ON messages (dialog_id, random_id) "
"WHERE random_id IS NOT NULL"));
TRY_STATUS(
db.exec("CREATE INDEX IF NOT EXISTS message_by_unique_message_id ON messages "
"(unique_message_id) WHERE unique_message_id IS NOT NULL"));
TRY_STATUS(
db.exec("CREATE INDEX IF NOT EXISTS message_by_ttl ON messages "
"(ttl_expires_at) WHERE ttl_expires_at IS NOT NULL"));
TRY_STATUS(add_media_indices(0, MESSAGES_DB_INDEX_COUNT));
TRY_STATUS(add_fts());
TRY_STATUS(add_call_index());
TRY_STATUS(add_notification_id_index());
TRY_STATUS(add_scheduled_messages_table());
version = current_db_version();
}
if (version < static_cast<int32>(DbVersion::MessagesDbMediaIndex)) {
TRY_STATUS(db.exec("ALTER TABLE messages ADD COLUMN index_mask INT4"));
TRY_STATUS(add_media_indices(0, MESSAGES_DB_INDEX_COUNT_OLD));
}
if (version < static_cast<int32>(DbVersion::MessagesDb30MediaIndex)) {
TRY_STATUS(add_media_indices(MESSAGES_DB_INDEX_COUNT_OLD, MESSAGES_DB_INDEX_COUNT));
}
if (version < static_cast<int32>(DbVersion::MessagesDbFts)) {
TRY_STATUS(db.exec("ALTER TABLE messages ADD COLUMN search_id INT8"));
TRY_STATUS(db.exec("ALTER TABLE messages ADD COLUMN text STRING"));
TRY_STATUS(add_fts());
}
if (version < static_cast<int32>(DbVersion::MessagesCallIndex)) {
TRY_STATUS(add_call_index());
}
if (version < static_cast<int32>(DbVersion::AddNotificationsSupport)) {
TRY_STATUS(db.exec("ALTER TABLE messages ADD COLUMN notification_id INT4"));
TRY_STATUS(add_notification_id_index());
}
if (version < static_cast<int32>(DbVersion::AddScheduledMessages)) {
TRY_STATUS(add_scheduled_messages_table());
}
return Status::OK();
}
// NB: must happen inside a transaction
Status drop_messages_db(SqliteDb &db, int32 version) {
LOG(WARNING) << "Drop message database " << tag("version", version)
<< tag("current_db_version", current_db_version());
return db.exec("DROP TABLE IF EXISTS messages");
}
class MessagesDbImpl : public MessagesDbSyncInterface {
public:
explicit MessagesDbImpl(SqliteDb db) : db_(std::move(db)) {
init().ensure();
}
Status init() {
TRY_RESULT_ASSIGN(
add_message_stmt_,
db_.get_statement("INSERT OR REPLACE INTO messages VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)"));
TRY_RESULT_ASSIGN(delete_message_stmt_,
db_.get_statement("DELETE FROM messages WHERE dialog_id = ?1 AND message_id = ?2"));
TRY_RESULT_ASSIGN(delete_all_dialog_messages_stmt_,
db_.get_statement("DELETE FROM messages WHERE dialog_id = ?1 AND message_id <= ?2"));
TRY_RESULT_ASSIGN(delete_dialog_messages_from_user_stmt_,
db_.get_statement("DELETE FROM messages WHERE dialog_id = ?1 AND sender_user_id == ?2"));
TRY_RESULT_ASSIGN(get_message_stmt_,
db_.get_statement("SELECT data FROM messages WHERE dialog_id = ?1 AND message_id = ?2"));
TRY_RESULT_ASSIGN(get_message_by_random_id_stmt_,
db_.get_statement("SELECT data FROM messages WHERE dialog_id = ?1 AND random_id = ?2"));
TRY_RESULT_ASSIGN(get_message_by_unique_message_id_stmt_,
db_.get_statement("SELECT dialog_id, data FROM messages WHERE unique_message_id = ?1"));
TRY_RESULT_ASSIGN(
get_expiring_messages_stmt_,
db_.get_statement("SELECT dialog_id, data FROM messages WHERE ?1 < ttl_expires_at AND ttl_expires_at <= ?2"));
TRY_RESULT_ASSIGN(get_expiring_messages_helper_stmt_,
db_.get_statement("SELECT MAX(ttl_expires_at), COUNT(*) FROM (SELECT ttl_expires_at FROM "
"messages WHERE ?1 < ttl_expires_at LIMIT ?2) AS T"));
TRY_RESULT_ASSIGN(get_messages_stmt_.asc_stmt_,
db_.get_statement("SELECT data, message_id FROM messages WHERE dialog_id = ?1 AND message_id > "
"?2 ORDER BY message_id ASC LIMIT ?3"));
TRY_RESULT_ASSIGN(get_messages_stmt_.desc_stmt_,
db_.get_statement("SELECT data, message_id FROM messages WHERE dialog_id = ?1 AND message_id < "
"?2 ORDER BY message_id DESC LIMIT ?3"));
TRY_RESULT_ASSIGN(get_scheduled_messages_stmt_,
db_.get_statement("SELECT data, message_id FROM scheduled_messages WHERE dialog_id = ?1 AND "
"message_id < ?2 ORDER BY message_id DESC LIMIT ?3"));
TRY_RESULT_ASSIGN(get_messages_from_notification_id_stmt_,
db_.get_statement("SELECT data, message_id FROM messages WHERE dialog_id = ?1 AND "
"notification_id < ?2 ORDER BY notification_id DESC LIMIT ?3"));
TRY_RESULT_ASSIGN(
get_messages_fts_stmt_,
db_.get_statement(
"SELECT dialog_id, data, search_id FROM messages WHERE search_id IN (SELECT rowid FROM messages_fts WHERE "
"messages_fts MATCH ?1 AND rowid < ?2 ORDER BY rowid DESC LIMIT ?3) ORDER BY search_id DESC"));
for (int32 i = 0; i < MESSAGES_DB_INDEX_COUNT; i++) {
TRY_RESULT_ASSIGN(get_messages_from_index_stmts_[i].desc_stmt_,
db_.get_statement(PSLICE() << "SELECT data, message_id FROM messages WHERE dialog_id = ?1 "
"AND message_id < ?2 AND (index_mask & "
<< (1 << i) << ") != 0 ORDER BY message_id DESC LIMIT ?3"));
TRY_RESULT_ASSIGN(get_messages_from_index_stmts_[i].asc_stmt_,
db_.get_statement(PSLICE() << "SELECT data, message_id FROM messages WHERE dialog_id = ?1 "
"AND message_id > ?2 AND (index_mask & "
<< (1 << i) << ") != 0 ORDER BY message_id ASC LIMIT ?3"));
// LOG(ERROR) << get_messages_from_index_stmts_[i].desc_stmt_.explain().ok();
// LOG(ERROR) << get_messages_from_index_stmts_[i].asc_stmt_.explain().ok();
}
for (int i = static_cast<int>(SearchMessagesFilter::Call) - 1, pos = 0;
i < static_cast<int>(SearchMessagesFilter::MissedCall); i++, pos++) {
TRY_RESULT_ASSIGN(
get_calls_stmts_[pos],
db_.get_statement(
PSLICE() << "SELECT dialog_id, data FROM messages WHERE unique_message_id < ?1 AND (index_mask & "
<< (1 << i) << ") != 0 ORDER BY unique_message_id DESC LIMIT ?2"));
}
TRY_RESULT_ASSIGN(add_scheduled_message_stmt_,
db_.get_statement("INSERT OR REPLACE INTO scheduled_messages VALUES(?1, ?2, ?3, ?4)"));
TRY_RESULT_ASSIGN(
get_scheduled_message_stmt_,
db_.get_statement("SELECT data FROM scheduled_messages WHERE dialog_id = ?1 AND message_id = ?2"));
TRY_RESULT_ASSIGN(
get_scheduled_server_message_stmt_,
db_.get_statement("SELECT data FROM scheduled_messages WHERE dialog_id = ?1 AND server_message_id = ?2"));
TRY_RESULT_ASSIGN(delete_scheduled_message_stmt_,
db_.get_statement("DELETE FROM scheduled_messages WHERE dialog_id = ?1 AND message_id = ?2"));
TRY_RESULT_ASSIGN(
delete_scheduled_server_message_stmt_,
db_.get_statement("DELETE FROM scheduled_messages WHERE dialog_id = ?1 AND server_message_id = ?2"));
// LOG(ERROR) << get_message_stmt_.explain().ok();
// LOG(ERROR) << get_messages_from_notification_id_stmt.explain().ok();
// LOG(ERROR) << get_message_by_random_id_stmt_.explain().ok();
// LOG(ERROR) << get_message_by_unique_message_id_stmt_.explain().ok();
// LOG(ERROR) << get_expiring_messages_stmt_.explain().ok();
// LOG(ERROR) << get_expiring_messages_helper_stmt_.explain().ok();
// LOG(FATAL) << "EXPLAINED";
return Status::OK();
}
Status add_message(FullMessageId full_message_id, ServerMessageId unique_message_id, UserId sender_user_id,
int64 random_id, int32 ttl_expires_at, int32 index_mask, int64 search_id, string text,
NotificationId notification_id, BufferSlice data) override {
LOG(INFO) << "Add " << full_message_id << " to database";
auto dialog_id = full_message_id.get_dialog_id();
auto message_id = full_message_id.get_message_id();
LOG_CHECK(dialog_id.is_valid()) << dialog_id << ' ' << message_id << ' ' << full_message_id;
CHECK(message_id.is_valid());
SCOPE_EXIT {
add_message_stmt_.reset();
};
add_message_stmt_.bind_int64(1, dialog_id.get()).ensure();
add_message_stmt_.bind_int64(2, message_id.get()).ensure();
if (unique_message_id.is_valid()) {
add_message_stmt_.bind_int32(3, unique_message_id.get()).ensure();
} else {
add_message_stmt_.bind_null(3).ensure();
}
if (sender_user_id.is_valid()) {
add_message_stmt_.bind_int32(4, sender_user_id.get()).ensure();
} else {
add_message_stmt_.bind_null(4).ensure();
}
if (random_id != 0) {
add_message_stmt_.bind_int64(5, random_id).ensure();
} else {
add_message_stmt_.bind_null(5).ensure();
}
add_message_stmt_.bind_blob(6, data.as_slice()).ensure();
if (ttl_expires_at != 0) {
add_message_stmt_.bind_int32(7, ttl_expires_at).ensure();
} else {
add_message_stmt_.bind_null(7).ensure();
}
if (index_mask != 0) {
add_message_stmt_.bind_int32(8, index_mask).ensure();
} else {
add_message_stmt_.bind_null(8).ensure();
}
if (search_id != 0) {
// add dialog_id to text
text += PSTRING() << " \a" << dialog_id.get();
if (index_mask) {
for (int i = 0; i < MESSAGES_DB_INDEX_COUNT; i++) {
if ((index_mask & (1 << i))) {
text += PSTRING() << " \a\a" << i;
}
}
}
add_message_stmt_.bind_int64(9, search_id).ensure();
} else {
text = "";
add_message_stmt_.bind_null(9).ensure();
}
if (!text.empty()) {
add_message_stmt_.bind_string(10, text).ensure();
} else {
add_message_stmt_.bind_null(10).ensure();
}
if (notification_id.is_valid()) {
add_message_stmt_.bind_int32(11, notification_id.get()).ensure();
} else {
add_message_stmt_.bind_null(11).ensure();
}
add_message_stmt_.step().ensure();
return Status::OK();
}
Status add_scheduled_message(FullMessageId full_message_id, BufferSlice data) override {
LOG(INFO) << "Add " << full_message_id << " to database";
auto dialog_id = full_message_id.get_dialog_id();
auto message_id = full_message_id.get_message_id();
CHECK(dialog_id.is_valid());
CHECK(message_id.is_valid_scheduled());
SCOPE_EXIT {
add_scheduled_message_stmt_.reset();
};
add_scheduled_message_stmt_.bind_int64(1, dialog_id.get()).ensure();
add_scheduled_message_stmt_.bind_int64(2, message_id.get()).ensure();
if (message_id.is_scheduled_server()) {
add_scheduled_message_stmt_.bind_int32(3, message_id.get_scheduled_server_message_id().get()).ensure();
} else {
add_scheduled_message_stmt_.bind_null(3).ensure();
}
add_scheduled_message_stmt_.bind_blob(4, data.as_slice()).ensure();
add_scheduled_message_stmt_.step().ensure();
return Status::OK();
}
Status delete_message(FullMessageId full_message_id) override {
LOG(INFO) << "Delete " << full_message_id << " from database";
auto dialog_id = full_message_id.get_dialog_id();
auto message_id = full_message_id.get_message_id();
CHECK(dialog_id.is_valid());
CHECK(message_id.is_valid() || message_id.is_valid_scheduled());
bool is_scheduled = message_id.is_scheduled();
bool is_scheduled_server = is_scheduled && message_id.is_scheduled_server();
auto &stmt = is_scheduled
? (is_scheduled_server ? delete_scheduled_server_message_stmt_ : delete_scheduled_message_stmt_)
: delete_message_stmt_;
SCOPE_EXIT {
stmt.reset();
};
stmt.bind_int64(1, dialog_id.get()).ensure();
if (is_scheduled_server) {
stmt.bind_int32(2, message_id.get_scheduled_server_message_id().get()).ensure();
} else {
stmt.bind_int64(2, message_id.get()).ensure();
}
stmt.step().ensure();
return Status::OK();
}
Status delete_all_dialog_messages(DialogId dialog_id, MessageId from_message_id) override {
LOG(INFO) << "Delete all messages in " << dialog_id << " up to " << from_message_id << " from database";
CHECK(dialog_id.is_valid());
CHECK(from_message_id.is_valid());
SCOPE_EXIT {
delete_all_dialog_messages_stmt_.reset();
};
delete_all_dialog_messages_stmt_.bind_int64(1, dialog_id.get()).ensure();
delete_all_dialog_messages_stmt_.bind_int64(2, from_message_id.get()).ensure();
auto status = delete_all_dialog_messages_stmt_.step();
if (status.is_error()) {
LOG(ERROR) << status;
}
return status;
}
Status delete_dialog_messages_from_user(DialogId dialog_id, UserId sender_user_id) override {
LOG(INFO) << "Delete all messages in " << dialog_id << " sent by " << sender_user_id << " from database";
CHECK(dialog_id.is_valid());
CHECK(sender_user_id.is_valid());
SCOPE_EXIT {
delete_dialog_messages_from_user_stmt_.reset();
};
delete_dialog_messages_from_user_stmt_.bind_int64(1, dialog_id.get()).ensure();
delete_dialog_messages_from_user_stmt_.bind_int32(2, sender_user_id.get()).ensure();
delete_dialog_messages_from_user_stmt_.step().ensure();
return Status::OK();
}
Result<BufferSlice> get_message(FullMessageId full_message_id) override {
auto dialog_id = full_message_id.get_dialog_id();
auto message_id = full_message_id.get_message_id();
CHECK(dialog_id.is_valid());
CHECK(message_id.is_valid() || message_id.is_valid_scheduled());
bool is_scheduled = message_id.is_scheduled();
bool is_scheduled_server = is_scheduled && message_id.is_scheduled_server();
auto &stmt = is_scheduled ? (is_scheduled_server ? get_scheduled_server_message_stmt_ : get_scheduled_message_stmt_)
: get_message_stmt_;
SCOPE_EXIT {
stmt.reset();
};
stmt.bind_int64(1, dialog_id.get()).ensure();
if (is_scheduled_server) {
stmt.bind_int32(2, message_id.get_scheduled_server_message_id().get()).ensure();
} else {
stmt.bind_int64(2, message_id.get()).ensure();
}
stmt.step().ensure();
if (!stmt.has_row()) {
return Status::Error("Not found");
}
return BufferSlice(stmt.view_blob(0));
}
Result<std::pair<DialogId, BufferSlice>> get_message_by_unique_message_id(
ServerMessageId unique_message_id) override {
if (!unique_message_id.is_valid()) {
return Status::Error("Invalid unique_message_id");
}
SCOPE_EXIT {
get_message_by_unique_message_id_stmt_.reset();
};
get_message_by_unique_message_id_stmt_.bind_int32(1, unique_message_id.get()).ensure();
get_message_by_unique_message_id_stmt_.step().ensure();
if (!get_message_by_unique_message_id_stmt_.has_row()) {
return Status::Error("Not found");
}
DialogId dialog_id(get_message_by_unique_message_id_stmt_.view_int64(0));
return std::make_pair(dialog_id, BufferSlice(get_message_by_unique_message_id_stmt_.view_blob(1)));
}
Result<BufferSlice> get_message_by_random_id(DialogId dialog_id, int64 random_id) override {
SCOPE_EXIT {
get_message_by_random_id_stmt_.reset();
};
get_message_by_random_id_stmt_.bind_int64(1, dialog_id.get()).ensure();
get_message_by_random_id_stmt_.bind_int64(2, random_id).ensure();
get_message_by_random_id_stmt_.step().ensure();
if (!get_message_by_random_id_stmt_.has_row()) {
return Status::Error("Not found");
}
return BufferSlice(get_message_by_random_id_stmt_.view_blob(0));
}
Result<BufferSlice> get_dialog_message_by_date(DialogId dialog_id, MessageId first_message_id,
MessageId last_message_id, int32 date) override {
int64 left_message_id = first_message_id.get();
int64 right_message_id = last_message_id.get();
LOG_CHECK(left_message_id <= right_message_id) << first_message_id << " " << last_message_id;
TRY_RESULT(first_messages, get_messages_inner(get_messages_stmt_.asc_stmt_, dialog_id, left_message_id - 1, 1));
if (!first_messages.empty()) {
MessageId real_first_message_id;
int32 real_first_message_date;
std::tie(real_first_message_id, real_first_message_date) = get_message_info(first_messages[0]);
if (real_first_message_date <= date) {
// we definitely have at least one suitable message, let's do a binary search
left_message_id = real_first_message_id.get();
MessageId prev_found_message_id;
while (left_message_id <= right_message_id) {
auto middle_message_id = left_message_id + ((right_message_id - left_message_id) >> 1);
TRY_RESULT(messages, get_messages_inner(get_messages_stmt_.asc_stmt_, dialog_id, middle_message_id, 1));
MessageId message_id;
int32 message_date = std::numeric_limits<int32>::max();
if (!messages.empty()) {
std::tie(message_id, message_date) = get_message_info(messages[0]);
}
if (message_date <= date) {
left_message_id = message_id.get();
} else {
right_message_id = middle_message_id - 1;
}
if (prev_found_message_id == message_id) {
// we may be very close to the result, let's check
TRY_RESULT(left_messages,
get_messages_inner(get_messages_stmt_.asc_stmt_, dialog_id, left_message_id - 1, 2));
CHECK(!left_messages.empty());
if (left_messages.size() == 1) {
// only one message has left, result is found
break;
}
MessageId next_message_id;
int32 next_message_date;
std::tie(next_message_id, next_message_date) = get_message_info(left_messages[1]);
if (next_message_date <= date) {
// next message has lesser date, adjusting left message
left_message_id = next_message_id.get();
} else {
// next message has bigger date, result is found
break;
}
}
prev_found_message_id = message_id;
}
// left_message_id is always an id of suitable message, let's return it
return get_message({dialog_id, MessageId(left_message_id)});
}
}
return Status::Error("Not found");
}
Result<std::pair<std::vector<std::pair<DialogId, BufferSlice>>, int32>> get_expiring_messages(int32 expires_from,
int32 expires_till,
int32 limit) override {
SCOPE_EXIT {
get_expiring_messages_stmt_.reset();
get_expiring_messages_helper_stmt_.reset();
};
std::vector<std::pair<DialogId, BufferSlice>> messages;
// load messages
if (expires_from <= expires_till) {
get_expiring_messages_stmt_.bind_int32(1, expires_from).ensure();
get_expiring_messages_stmt_.bind_int32(2, expires_till).ensure();
get_expiring_messages_stmt_.step().ensure();
while (get_expiring_messages_stmt_.has_row()) {
DialogId dialog_id(get_expiring_messages_stmt_.view_int64(0));
BufferSlice data(get_expiring_messages_stmt_.view_blob(1));
messages.push_back(std::make_pair(dialog_id, std::move(data)));
get_expiring_messages_stmt_.step().ensure();
}
}
// calc next expires_till
get_expiring_messages_helper_stmt_.bind_int32(1, expires_till).ensure();
get_expiring_messages_helper_stmt_.bind_int32(2, limit).ensure();
get_expiring_messages_helper_stmt_.step().ensure();
CHECK(get_expiring_messages_helper_stmt_.has_row());
int32 count = get_expiring_messages_helper_stmt_.view_int32(1);
int32 next_expires_till = -1;
if (count != 0) {
next_expires_till = get_expiring_messages_helper_stmt_.view_int32(0);
}
return std::make_pair(std::move(messages), next_expires_till);
}
Result<std::vector<BufferSlice>> get_messages(MessagesDbMessagesQuery query) override {
if (query.index_mask != 0) {
return get_messages_from_index(query.dialog_id, query.from_message_id, query.index_mask, query.offset,
query.limit);
}
return get_messages_impl(get_messages_stmt_, query.dialog_id, query.from_message_id, query.offset, query.limit);
}
Result<std::vector<BufferSlice>> get_scheduled_messages(DialogId dialog_id, int32 limit) override {
return get_messages_inner(get_scheduled_messages_stmt_, dialog_id, std::numeric_limits<int64>::max(), limit);
}
Result<vector<BufferSlice>> get_messages_from_notification_id(DialogId dialog_id, NotificationId from_notification_id,
int32 limit) override {
auto &stmt = get_messages_from_notification_id_stmt_;
SCOPE_EXIT {
stmt.reset();
};
stmt.bind_int64(1, dialog_id.get()).ensure();
stmt.bind_int32(2, from_notification_id.get()).ensure();
stmt.bind_int32(3, limit).ensure();
std::vector<BufferSlice> result;
stmt.step().ensure();
while (stmt.has_row()) {
auto data_slice = stmt.view_blob(0);
result.emplace_back(data_slice);
auto message_id = stmt.view_int64(1);
LOG(INFO) << "Load " << MessageId(message_id) << " in " << dialog_id << " from database";
stmt.step().ensure();
}
return std::move(result);
}
static string prepare_query(Slice query) {
auto is_word_character = [](uint32 a) {
switch (get_unicode_simple_category(a)) {
case UnicodeSimpleCategory::Letter:
case UnicodeSimpleCategory::DecimalNumber:
case UnicodeSimpleCategory::Number:
return true;
default:
return a == '_';
}
};
const size_t MAX_QUERY_SIZE = 1024;
query = utf8_truncate(query, MAX_QUERY_SIZE);
auto buf = StackAllocator::alloc(query.size() * 4 + 100);
StringBuilder sb(buf.as_slice());
bool in_word{false};
for (auto ptr = query.ubegin(), end = query.uend(); ptr < end;) {
uint32 code;
auto code_ptr = ptr;
ptr = next_utf8_unsafe(ptr, &code, "prepare_query");
if (is_word_character(code)) {
if (!in_word) {
in_word = true;
sb << "\"";
}
sb << Slice(code_ptr, ptr);
} else {
if (in_word) {
in_word = false;
sb << "\" ";
}
}
}
if (in_word) {
sb << "\" ";
}
if (sb.is_error()) {
LOG(ERROR) << "StringBuilder buffer overflow";
return "";
}
return sb.as_cslice().str();
}
Result<MessagesDbFtsResult> get_messages_fts(MessagesDbFtsQuery query) override {
SCOPE_EXIT {
get_messages_fts_stmt_.reset();
};
LOG(INFO) << tag("query", query.query) << query.dialog_id << tag("index_mask", query.index_mask)
<< tag("from_search_id", query.from_search_id) << tag("limit", query.limit);
string words = prepare_query(query.query);
LOG(INFO) << tag("from", query.query) << tag("to", words);
// dialog_id kludge
if (query.dialog_id.is_valid()) {
words += PSTRING() << " \"\a" << query.dialog_id.get() << "\"";
}
// index_mask kludge
if (query.index_mask != 0) {
int index_i = -1;
for (int i = 0; i < MESSAGES_DB_INDEX_COUNT; i++) {
if (query.index_mask == (1 << i)) {
index_i = i;
break;
}
}
if (index_i == -1) {
return Status::Error("Union of index types is not supported");
}
words += PSTRING() << " \"\a\a" << index_i << "\"";
}
auto &stmt = get_messages_fts_stmt_;
stmt.bind_string(1, words).ensure();
if (query.from_search_id == 0) {
query.from_search_id = std::numeric_limits<int64>::max();
}
stmt.bind_int64(2, query.from_search_id).ensure();
stmt.bind_int32(3, query.limit).ensure();
MessagesDbFtsResult result;
auto status = stmt.step();
if (status.is_error()) {
LOG(ERROR) << status;
return std::move(result);
}
while (stmt.has_row()) {
auto dialog_id = stmt.view_int64(0);
auto data_slice = stmt.view_blob(1);
auto search_id = stmt.view_int64(2);
result.next_search_id = search_id;
result.messages.push_back(MessagesDbMessage{DialogId(dialog_id), BufferSlice(data_slice)});
stmt.step().ensure();
}
return std::move(result);
}
Result<std::vector<BufferSlice>> get_messages_from_index(DialogId dialog_id, MessageId from_message_id,
int32 index_mask, int32 offset, int32 limit) {
CHECK(index_mask != 0);
LOG_CHECK(index_mask < (1 << MESSAGES_DB_INDEX_COUNT)) << tag("index_mask", index_mask);
int index_i = -1;
for (int i = 0; i < MESSAGES_DB_INDEX_COUNT; i++) {
if (index_mask == (1 << i)) {
index_i = i;
break;
}
}
if (index_i == -1) {
return Status::Error("Union is not supported");
}
auto &stmt = get_messages_from_index_stmts_[index_i];
return get_messages_impl(stmt, dialog_id, from_message_id, offset, limit);
}
Result<MessagesDbCallsResult> get_calls(MessagesDbCallsQuery query) override {
CHECK(query.index_mask != 0);
LOG_CHECK(query.index_mask < (1 << MESSAGES_DB_INDEX_COUNT)) << tag("index_mask", query.index_mask);
int index_i = -1;
for (int i = 0; i < MESSAGES_DB_INDEX_COUNT; i++) {
if (query.index_mask == (1 << i)) {
index_i = i;
break;
}
}
if (index_i == -1) {
return Status::Error("Union is not supported");
}
int32 pos;
if (index_i + 1 == static_cast<int>(SearchMessagesFilter::Call)) {
pos = 0;
} else if (index_i + 1 == static_cast<int>(SearchMessagesFilter::MissedCall)) {
pos = 1;
} else {
return Status::Error(PSLICE() << "Index_mask is not Call or MissedCall " << query.index_mask);
}
auto &stmt = get_calls_stmts_[pos];
SCOPE_EXIT {
stmt.reset();
};
stmt.bind_int32(1, query.from_unique_message_id).ensure();
stmt.bind_int32(2, query.limit).ensure();
MessagesDbCallsResult result;
stmt.step().ensure();
while (stmt.has_row()) {
auto dialog_id = stmt.view_int64(0);
auto data_slice = stmt.view_blob(1);
result.messages.push_back(MessagesDbMessage{DialogId(dialog_id), BufferSlice(data_slice)});
stmt.step().ensure();
}
return std::move(result);
}
Status begin_transaction() override {
return db_.begin_transaction();
}
Status commit_transaction() override {
return db_.commit_transaction();
}
private:
SqliteDb db_;
SqliteStatement add_message_stmt_;
SqliteStatement delete_message_stmt_;
SqliteStatement delete_all_dialog_messages_stmt_;
SqliteStatement delete_dialog_messages_from_user_stmt_;
SqliteStatement get_message_stmt_;
SqliteStatement get_message_by_random_id_stmt_;
SqliteStatement get_message_by_unique_message_id_stmt_;
SqliteStatement get_expiring_messages_stmt_;
SqliteStatement get_expiring_messages_helper_stmt_;
struct GetMessagesStmt {
SqliteStatement asc_stmt_;
SqliteStatement desc_stmt_;
};
GetMessagesStmt get_messages_stmt_;
SqliteStatement get_scheduled_messages_stmt_;
SqliteStatement get_messages_from_notification_id_stmt_;
std::array<GetMessagesStmt, MESSAGES_DB_INDEX_COUNT> get_messages_from_index_stmts_;
std::array<SqliteStatement, 2> get_calls_stmts_;
SqliteStatement get_messages_fts_stmt_;
SqliteStatement add_scheduled_message_stmt_;
SqliteStatement get_scheduled_message_stmt_;
SqliteStatement get_scheduled_server_message_stmt_;
SqliteStatement delete_scheduled_message_stmt_;
SqliteStatement delete_scheduled_server_message_stmt_;
Result<std::vector<BufferSlice>> get_messages_impl(GetMessagesStmt &stmt, DialogId dialog_id,
MessageId from_message_id, int32 offset, int32 limit) {
LOG_CHECK(dialog_id.is_valid()) << dialog_id;
CHECK(from_message_id.is_valid());
LOG(INFO) << "Loading messages in " << dialog_id << " from " << from_message_id << " with offset = " << offset
<< " and limit = " << limit;
auto message_id = from_message_id.get();
if (message_id >= MessageId::max().get()) {
message_id--;
}
auto left_message_id = message_id;
auto left_cnt = limit + offset;
auto right_message_id = message_id - 1;
auto right_cnt = -offset;
std::vector<BufferSlice> left;
std::vector<BufferSlice> right;
if (left_cnt != 0) {
if (right_cnt == 1 && false) {
left_message_id++;
left_cnt++;
}
TRY_RESULT(left_tmp, get_messages_inner(stmt.desc_stmt_, dialog_id, left_message_id, left_cnt));
left = std::move(left_tmp);
if (right_cnt == 1 && !left.empty() && false /*get_message_id(left[0].as_slice()) == message_id*/) {
right_cnt = 0;
}
}
if (right_cnt != 0) {
TRY_RESULT(right_tmp, get_messages_inner(stmt.asc_stmt_, dialog_id, right_message_id, right_cnt));
right = std::move(right_tmp);
std::reverse(right.begin(), right.end());
}
if (left.empty()) {
return std::move(right);
}
if (right.empty()) {
return std::move(left);
}
right.reserve(right.size() + left.size());
std::move(left.begin(), left.end(), std::back_inserter(right));
return std::move(right);
}
Result<std::vector<BufferSlice>> get_messages_inner(SqliteStatement &stmt, DialogId dialog_id, int64 from_message_id,
int32 limit) {
SCOPE_EXIT {
stmt.reset();
};
stmt.bind_int64(1, dialog_id.get()).ensure();
stmt.bind_int64(2, from_message_id).ensure();
stmt.bind_int32(3, limit).ensure();
LOG(INFO) << "Begin to load " << limit << " messages in " << dialog_id << " from " << MessageId(from_message_id)
<< " from database";
std::vector<BufferSlice> result;
stmt.step().ensure();
while (stmt.has_row()) {
auto data_slice = stmt.view_blob(0);
result.emplace_back(data_slice);
auto message_id = stmt.view_int64(1);
LOG(INFO) << "Loaded " << MessageId(message_id) << " in " << dialog_id << " from database";
stmt.step().ensure();
}
return std::move(result);
}
static std::tuple<MessageId, int32> get_message_info(const BufferSlice &message) {
LogEventParser message_date_parser(message.as_slice());
int32 flags;
td::parse(flags, message_date_parser);
int32 flags2 = 0;
if ((flags & (1 << 29)) != 0) {
td::parse(flags2, message_date_parser);
}
bool has_sender = (flags >> 10) & 1;
MessageId message_id;
td::parse(message_id, message_date_parser);
UserId sender_user_id;
if (has_sender) {
td::parse(sender_user_id, message_date_parser);
}
int32 date;
td::parse(date, message_date_parser);
LOG(INFO) << "Loaded " << message_id << " sent at " << date << " by " << sender_user_id;
return std::make_tuple(message_id, date);
}
};
std::shared_ptr<MessagesDbSyncSafeInterface> create_messages_db_sync(
std::shared_ptr<SqliteConnectionSafe> sqlite_connection) {
class MessagesDbSyncSafe : public MessagesDbSyncSafeInterface {
public:
explicit MessagesDbSyncSafe(std::shared_ptr<SqliteConnectionSafe> sqlite_connection)
: lsls_db_([safe_connection = std::move(sqlite_connection)] {
return make_unique<MessagesDbImpl>(safe_connection->get().clone());
}) {
}
MessagesDbSyncInterface &get() override {
return *lsls_db_.get();
}
private:
LazySchedulerLocalStorage<unique_ptr<MessagesDbSyncInterface>> lsls_db_;
};
return std::make_shared<MessagesDbSyncSafe>(std::move(sqlite_connection));
}
class MessagesDbAsync : public MessagesDbAsyncInterface {
public:
MessagesDbAsync(std::shared_ptr<MessagesDbSyncSafeInterface> sync_db, int32 scheduler_id) {
impl_ = create_actor_on_scheduler<Impl>("MessagesDbActor", scheduler_id, std::move(sync_db));
}
void add_message(FullMessageId full_message_id, ServerMessageId unique_message_id, UserId sender_user_id,
int64 random_id, int32 ttl_expires_at, int32 index_mask, int64 search_id, string text,
NotificationId notification_id, BufferSlice data, Promise<> promise) override {
send_closure_later(impl_, &Impl::add_message, full_message_id, unique_message_id, sender_user_id, random_id,
ttl_expires_at, index_mask, search_id, std::move(text), notification_id, std::move(data),
std::move(promise));
}
void add_scheduled_message(FullMessageId full_message_id, BufferSlice data, Promise<> promise) override {
send_closure_later(impl_, &Impl::add_scheduled_message, full_message_id, std::move(data), std::move(promise));
}
void delete_message(FullMessageId full_message_id, Promise<> promise) override {
send_closure_later(impl_, &Impl::delete_message, full_message_id, std::move(promise));
}
void delete_all_dialog_messages(DialogId dialog_id, MessageId from_message_id, Promise<> promise) override {
send_closure_later(impl_, &Impl::delete_all_dialog_messages, dialog_id, from_message_id, std::move(promise));
}
void delete_dialog_messages_from_user(DialogId dialog_id, UserId sender_user_id, Promise<> promise) override {
send_closure_later(impl_, &Impl::delete_dialog_messages_from_user, dialog_id, sender_user_id, std::move(promise));
}
void get_message(FullMessageId full_message_id, Promise<BufferSlice> promise) override {
send_closure_later(impl_, &Impl::get_message, full_message_id, std::move(promise));
}
void get_message_by_unique_message_id(ServerMessageId unique_message_id,
Promise<std::pair<DialogId, BufferSlice>> promise) override {
send_closure_later(impl_, &Impl::get_message_by_unique_message_id, unique_message_id, std::move(promise));
}
void get_message_by_random_id(DialogId dialog_id, int64 random_id, Promise<BufferSlice> promise) override {
send_closure_later(impl_, &Impl::get_message_by_random_id, dialog_id, random_id, std::move(promise));
}
void get_dialog_message_by_date(DialogId dialog_id, MessageId first_message_id, MessageId last_message_id, int32 date,
Promise<BufferSlice> promise) override {
send_closure_later(impl_, &Impl::get_dialog_message_by_date, dialog_id, first_message_id, last_message_id, date,
std::move(promise));
}
void get_messages(MessagesDbMessagesQuery query, Promise<std::vector<BufferSlice>> promise) override {
send_closure_later(impl_, &Impl::get_messages, std::move(query), std::move(promise));
}
void get_scheduled_messages(DialogId dialog_id, int32 limit, Promise<std::vector<BufferSlice>> promise) override {
send_closure_later(impl_, &Impl::get_scheduled_messages, dialog_id, limit, std::move(promise));
}
void get_messages_from_notification_id(DialogId dialog_id, NotificationId from_notification_id, int32 limit,
Promise<vector<BufferSlice>> promise) override {
send_closure_later(impl_, &Impl::get_messages_from_notification_id, dialog_id, from_notification_id, limit,
std::move(promise));
}
void get_calls(MessagesDbCallsQuery query, Promise<MessagesDbCallsResult> promise) override {
send_closure_later(impl_, &Impl::get_calls, std::move(query), std::move(promise));
}
void get_messages_fts(MessagesDbFtsQuery query, Promise<MessagesDbFtsResult> promise) override {
send_closure_later(impl_, &Impl::get_messages_fts, std::move(query), std::move(promise));
}
void get_expiring_messages(
int32 expires_from, int32 expires_till, int32 limit,
Promise<std::pair<std::vector<std::pair<DialogId, BufferSlice>>, int32>> promise) override {
send_closure_later(impl_, &Impl::get_expiring_messages, expires_from, expires_till, limit, std::move(promise));
}
void close(Promise<> promise) override {
send_closure_later(impl_, &Impl::close, std::move(promise));
}
void force_flush() override {
send_closure_later(impl_, &Impl::force_flush);
}
private:
class Impl : public Actor {
public:
explicit Impl(std::shared_ptr<MessagesDbSyncSafeInterface> sync_db_safe) : sync_db_safe_(std::move(sync_db_safe)) {
}
void add_message(FullMessageId full_message_id, ServerMessageId unique_message_id, UserId sender_user_id,
int64 random_id, int32 ttl_expires_at, int32 index_mask, int64 search_id, string text,
NotificationId notification_id, BufferSlice data, Promise<> promise) {
add_write_query([=, promise = std::move(promise), data = std::move(data), text = std::move(text)](Unit) mutable {
this->on_write_result(
std::move(promise),
sync_db_->add_message(full_message_id, unique_message_id, sender_user_id, random_id, ttl_expires_at,
index_mask, search_id, std::move(text), notification_id, std::move(data)));
});
}
void add_scheduled_message(FullMessageId full_message_id, BufferSlice data, Promise<> promise) {
add_write_query([this, full_message_id, promise = std::move(promise), data = std::move(data)](Unit) mutable {
this->on_write_result(std::move(promise), sync_db_->add_scheduled_message(full_message_id, std::move(data)));
});
}
void delete_message(FullMessageId full_message_id, Promise<> promise) {
add_write_query([=, promise = std::move(promise)](Unit) mutable {
this->on_write_result(std::move(promise), sync_db_->delete_message(full_message_id));
});
}
void on_write_result(Promise<> promise, Status status) {
pending_write_results_.emplace_back(std::move(promise), std::move(status));
}
void delete_all_dialog_messages(DialogId dialog_id, MessageId from_message_id, Promise<> promise) {
add_read_query();
promise.set_result(sync_db_->delete_all_dialog_messages(dialog_id, from_message_id));
}
void delete_dialog_messages_from_user(DialogId dialog_id, UserId sender_user_id, Promise<> promise) {
add_read_query();
promise.set_result(sync_db_->delete_dialog_messages_from_user(dialog_id, sender_user_id));
}
void get_message(FullMessageId full_message_id, Promise<BufferSlice> promise) {
add_read_query();
promise.set_result(sync_db_->get_message(full_message_id));
}
void get_message_by_unique_message_id(ServerMessageId unique_message_id,
Promise<std::pair<DialogId, BufferSlice>> promise) {
add_read_query();
promise.set_result(sync_db_->get_message_by_unique_message_id(unique_message_id));
}
void get_message_by_random_id(DialogId dialog_id, int64 random_id, Promise<BufferSlice> promise) {
add_read_query();
promise.set_result(sync_db_->get_message_by_random_id(dialog_id, random_id));
}
void get_dialog_message_by_date(DialogId dialog_id, MessageId first_message_id, MessageId last_message_id,
int32 date, Promise<BufferSlice> promise) {
add_read_query();
promise.set_result(sync_db_->get_dialog_message_by_date(dialog_id, first_message_id, last_message_id, date));
}
void get_messages(MessagesDbMessagesQuery query, Promise<std::vector<BufferSlice>> promise) {
add_read_query();
promise.set_result(sync_db_->get_messages(std::move(query)));
}
void get_scheduled_messages(DialogId dialog_id, int32 limit, Promise<std::vector<BufferSlice>> promise) {
add_read_query();
promise.set_result(sync_db_->get_scheduled_messages(dialog_id, limit));
}
void get_messages_from_notification_id(DialogId dialog_id, NotificationId from_notification_id, int32 limit,
Promise<vector<BufferSlice>> promise) {
add_read_query();
promise.set_result(sync_db_->get_messages_from_notification_id(dialog_id, from_notification_id, limit));
}
void get_calls(MessagesDbCallsQuery query, Promise<MessagesDbCallsResult> promise) {
add_read_query();
promise.set_result(sync_db_->get_calls(std::move(query)));
}
void get_messages_fts(MessagesDbFtsQuery query, Promise<MessagesDbFtsResult> promise) {
add_read_query();
promise.set_result(sync_db_->get_messages_fts(std::move(query)));
}
void get_expiring_messages(int32 expires_from, int32 expires_till, int32 limit,
Promise<std::pair<std::vector<std::pair<DialogId, BufferSlice>>, int32>> promise) {
add_read_query();
promise.set_result(sync_db_->get_expiring_messages(expires_from, expires_till, limit));
}
void close(Promise<> promise) {
do_flush();
sync_db_safe_.reset();
sync_db_ = nullptr;
promise.set_value(Unit());
stop();
}
void force_flush() {
LOG(INFO) << "MessagesDb flushed";
do_flush();
}
private:
std::shared_ptr<MessagesDbSyncSafeInterface> sync_db_safe_;
MessagesDbSyncInterface *sync_db_ = nullptr;
static constexpr size_t MAX_PENDING_QUERIES_COUNT{50};
static constexpr double MAX_PENDING_QUERIES_DELAY{0.01};
//NB: order is important, destructor of pending_writes_ will change pending_write_results_
std::vector<std::pair<Promise<>, Status>> pending_write_results_;
std::vector<Promise<>> pending_writes_;
double wakeup_at_ = 0;
template <class F>
void add_write_query(F &&f) {
pending_writes_.push_back(PromiseCreator::lambda(std::forward<F>(f), PromiseCreator::Ignore()));
if (pending_writes_.size() > MAX_PENDING_QUERIES_COUNT) {
do_flush();
wakeup_at_ = 0;
} else if (wakeup_at_ == 0) {
wakeup_at_ = Time::now_cached() + MAX_PENDING_QUERIES_DELAY;
}
if (wakeup_at_ != 0) {
set_timeout_at(wakeup_at_);
}
}
void add_read_query() {
do_flush();
}
void do_flush() {
if (pending_writes_.empty()) {
return;
}
sync_db_->begin_transaction().ensure();
for (auto &query : pending_writes_) {
query.set_value(Unit());
}
sync_db_->commit_transaction().ensure();
pending_writes_.clear();
for (auto &p : pending_write_results_) {
p.first.set_result(std::move(p.second));
}
pending_write_results_.clear();
cancel_timeout();
}
void timeout_expired() override {
do_flush();
}
void start_up() override {
sync_db_ = &sync_db_safe_->get();
}
};
ActorOwn<Impl> impl_;
};
std::shared_ptr<MessagesDbAsyncInterface> create_messages_db_async(std::shared_ptr<MessagesDbSyncSafeInterface> sync_db,
int32 scheduler_id) {
return std::make_shared<MessagesDbAsync>(std::move(sync_db), scheduler_id);
}
} // namespace td
Reference in New Issue View Git Blame Copy Permalink