Merge pull request #2136 from profanity-im/db-dup
refactor(database): enforce unique archive_id for reliable deduplication
This commit is contained in:
@@ -93,7 +93,7 @@ endif
|
||||
glib_dep = dependency('glib-2.0', version: '>= 2.62.0')
|
||||
gio_dep = dependency('gio-2.0')
|
||||
curl_dep = dependency('libcurl', version: '>= 7.62.0')
|
||||
sqlite_dep = dependency('sqlite3', version: '>= 3.22.0')
|
||||
sqlite_dep = dependency('sqlite3', version: '>= 3.35.0')
|
||||
thread_dep = dependency('threads')
|
||||
math_dep = cc.find_library('m')
|
||||
libstrophe_dep = dependency('libstrophe', version: '>= 0.12.3')
|
||||
|
||||
140
src/database.c
140
src/database.c
@@ -35,9 +35,10 @@ static prof_msg_type_t _get_message_type_type(const char* const type);
|
||||
static prof_enc_t _get_message_enc_type(const char* const encstr);
|
||||
static int _get_db_version(void);
|
||||
static gboolean _migrate_to_v2(void);
|
||||
static gboolean _migrate_to_v3(void);
|
||||
static gboolean _check_available_space_for_db_migration(char* path_to_db);
|
||||
|
||||
static const int latest_version = 2;
|
||||
static const int latest_version = 3;
|
||||
|
||||
static char*
|
||||
_db_strdup(const char* str)
|
||||
@@ -117,7 +118,7 @@ log_database_init(ProfAccount* account)
|
||||
"`timestamp` TEXT, "
|
||||
"`type` TEXT, "
|
||||
"`stanza_id` TEXT, "
|
||||
"`archive_id` TEXT, "
|
||||
"`archive_id` TEXT UNIQUE, "
|
||||
"`encryption` TEXT, "
|
||||
"`marked_read` INTEGER, "
|
||||
"`replace_id` TEXT, "
|
||||
@@ -158,7 +159,7 @@ log_database_init(ProfAccount* account)
|
||||
}
|
||||
|
||||
if (db_version == -1) {
|
||||
query = "INSERT OR IGNORE INTO `DbVersion` (`version`) VALUES ('2')";
|
||||
query = "INSERT INTO `DbVersion` (`version`) VALUES ('2') ON CONFLICT(`version`) DO NOTHING";
|
||||
if (SQLITE_OK != sqlite3_exec(g_chatlog_database, query, NULL, 0, &err_msg)) {
|
||||
goto out;
|
||||
}
|
||||
@@ -177,6 +178,10 @@ log_database_init(ProfAccount* account)
|
||||
cons_show_error("Database Initialization Error: Unable to migrate database to version 2. Please, check error logs for details.");
|
||||
goto out;
|
||||
}
|
||||
if (db_version < 3 && (!_check_available_space_for_db_migration(filename) || !_migrate_to_v3())) {
|
||||
cons_show_error("Database Initialization Error: Unable to migrate database to version 3. Please, check error logs for details.");
|
||||
goto out;
|
||||
}
|
||||
cons_show("Database schema migration was successful.");
|
||||
}
|
||||
|
||||
@@ -490,27 +495,7 @@ _add_to_db(ProfMessage* message, char* type, const Jid* const from_jid, const Ji
|
||||
}
|
||||
|
||||
// stanza-id (XEP-0359) doesn't have to be present in the message.
|
||||
// But if it's duplicated, it's a serious server-side problem, so we better track it.
|
||||
// Unless it's MAM, in that case it's expected behaviour.
|
||||
if (message->stanzaid && !message->is_mam) {
|
||||
auto_sqlite char* duplicate_check_query = sqlite3_mprintf("SELECT 1 FROM `ChatLogs` WHERE (`archive_id` = %Q)",
|
||||
message->stanzaid);
|
||||
|
||||
if (!duplicate_check_query) {
|
||||
log_error("Could not allocate memory for SQL duplicate query in log_database_add()");
|
||||
return;
|
||||
}
|
||||
|
||||
sqlite3_stmt* stmt;
|
||||
|
||||
if (SQLITE_OK == sqlite3_prepare_v2(g_chatlog_database, duplicate_check_query, -1, &stmt, NULL)) {
|
||||
if (sqlite3_step(stmt) == SQLITE_ROW) {
|
||||
log_error("Duplicate stanza-id found for the message. stanza_id: %s; archive_id: %s; sender: %s; content: %s", message->id, message->stanzaid, from_jid->barejid, message->plain);
|
||||
cons_show_error("Got a message with duplicate (server-generated) stanza-id from %s.", from_jid->fulljid);
|
||||
}
|
||||
sqlite3_finalize(stmt);
|
||||
}
|
||||
}
|
||||
// We use archive_id UNIQUE constraint and ON CONFLICT DO NOTHING for deduplication.
|
||||
|
||||
auto_sqlite char* orig_message_id = original_message_id == -1 ? NULL : sqlite3_mprintf("%d", original_message_id);
|
||||
|
||||
@@ -518,7 +503,9 @@ _add_to_db(ProfMessage* message, char* type, const Jid* const from_jid, const Ji
|
||||
"(`from_jid`, `from_resource`, `to_jid`, `to_resource`, "
|
||||
"`message`, `timestamp`, `stanza_id`, `archive_id`, "
|
||||
"`replaces_db_id`, `replace_id`, `type`, `encryption`) "
|
||||
"VALUES (%Q, %Q, %Q, %Q, %Q, %Q, %Q, %Q, %Q, %Q, %Q, %Q)",
|
||||
"VALUES (%Q, %Q, %Q, %Q, %Q, %Q, %Q, %Q, %Q, %Q, %Q, %Q) "
|
||||
"ON CONFLICT(`archive_id`) DO NOTHING "
|
||||
"RETURNING id",
|
||||
from_jid->barejid,
|
||||
from_jid->resourcepart,
|
||||
to_jid->barejid,
|
||||
@@ -538,19 +525,23 @@ _add_to_db(ProfMessage* message, char* type, const Jid* const from_jid, const Ji
|
||||
|
||||
log_debug("Writing to DB. Query: %s", query);
|
||||
|
||||
if (SQLITE_OK != sqlite3_exec(g_chatlog_database, query, NULL, 0, &err_msg)) {
|
||||
if (err_msg) {
|
||||
log_error("SQLite error in _add_to_db(): %s", err_msg);
|
||||
sqlite3_free(err_msg);
|
||||
} else {
|
||||
log_error("Unknown SQLite error in _add_to_db().");
|
||||
}
|
||||
} else {
|
||||
int inserted_rows_count = sqlite3_changes(g_chatlog_database);
|
||||
if (inserted_rows_count < 1) {
|
||||
log_error("SQLite did not insert message (rows: %d, id: %s, content: %s)", inserted_rows_count, message->id, message->plain);
|
||||
}
|
||||
sqlite3_stmt* stmt = NULL;
|
||||
int rc = sqlite3_prepare_v2(g_chatlog_database, query, -1, &stmt, NULL);
|
||||
if (rc != SQLITE_OK) {
|
||||
log_error("SQLite error in _add_to_db() (prepare): %s", sqlite3_errmsg(g_chatlog_database));
|
||||
return;
|
||||
}
|
||||
|
||||
rc = sqlite3_step(stmt);
|
||||
if (rc == SQLITE_ROW) {
|
||||
log_debug("Successfully inserted message into database.");
|
||||
} else if (rc == SQLITE_DONE) {
|
||||
log_debug("Message already exists in database (archive_id: %s), skipping.", message->stanzaid);
|
||||
} else {
|
||||
log_error("SQLite error in _add_to_db() (step): %s", sqlite3_errmsg(g_chatlog_database));
|
||||
}
|
||||
|
||||
sqlite3_finalize(stmt);
|
||||
}
|
||||
|
||||
static int
|
||||
@@ -656,3 +647,78 @@ _check_available_space_for_db_migration(char* path_to_db)
|
||||
return FALSE;
|
||||
}
|
||||
}
|
||||
|
||||
static gboolean
|
||||
_migrate_to_v3(void)
|
||||
{
|
||||
char* err_msg = NULL;
|
||||
|
||||
const char* sql_statements[] = {
|
||||
"BEGIN TRANSACTION",
|
||||
"CREATE TABLE `ChatLogs_v3_migration` ("
|
||||
"`id` INTEGER PRIMARY KEY AUTOINCREMENT, "
|
||||
"`from_jid` TEXT NOT NULL, "
|
||||
"`to_jid` TEXT NOT NULL, "
|
||||
"`from_resource` TEXT, "
|
||||
"`to_resource` TEXT, "
|
||||
"`message` TEXT, "
|
||||
"`timestamp` TEXT, "
|
||||
"`type` TEXT, "
|
||||
"`stanza_id` TEXT, "
|
||||
"`archive_id` TEXT UNIQUE, "
|
||||
"`encryption` TEXT, "
|
||||
"`marked_read` INTEGER, "
|
||||
"`replace_id` TEXT, "
|
||||
"`replaces_db_id` INTEGER, "
|
||||
"`replaced_by_db_id` INTEGER)",
|
||||
|
||||
"INSERT INTO `ChatLogs_v3_migration` "
|
||||
"SELECT * FROM `ChatLogs` "
|
||||
"WHERE `archive_id` IS NULL "
|
||||
"UNION ALL "
|
||||
"SELECT * FROM (SELECT * FROM `ChatLogs` WHERE `archive_id` IS NOT NULL GROUP BY `archive_id`)",
|
||||
|
||||
"DROP TABLE `ChatLogs` ",
|
||||
"ALTER TABLE `ChatLogs_v3_migration` RENAME TO `ChatLogs` ",
|
||||
|
||||
"CREATE INDEX ChatLogs_timestamp_IDX ON `ChatLogs` (`timestamp`)",
|
||||
"CREATE INDEX ChatLogs_to_from_jid_IDX ON `ChatLogs` (`to_jid`, `from_jid`)",
|
||||
"CREATE TRIGGER update_corrected_message "
|
||||
"AFTER INSERT ON ChatLogs "
|
||||
"FOR EACH ROW "
|
||||
"WHEN NEW.replaces_db_id IS NOT NULL "
|
||||
"BEGIN "
|
||||
"UPDATE ChatLogs "
|
||||
"SET replaced_by_db_id = NEW.id "
|
||||
"WHERE id = NEW.replaces_db_id; "
|
||||
"END;",
|
||||
|
||||
"UPDATE `DbVersion` SET `version` = 3;",
|
||||
"END TRANSACTION"
|
||||
};
|
||||
|
||||
int statements_count = sizeof(sql_statements) / sizeof(sql_statements[0]);
|
||||
|
||||
for (int i = 0; i < statements_count; i++) {
|
||||
if (SQLITE_OK != sqlite3_exec(g_chatlog_database, sql_statements[i], NULL, 0, &err_msg)) {
|
||||
log_error("SQLite error in _migrate_to_v3() on statement %d: %s", i, err_msg);
|
||||
if (err_msg) {
|
||||
sqlite3_free(err_msg);
|
||||
err_msg = NULL;
|
||||
}
|
||||
goto cleanup;
|
||||
}
|
||||
}
|
||||
|
||||
return TRUE;
|
||||
|
||||
cleanup:
|
||||
if (SQLITE_OK != sqlite3_exec(g_chatlog_database, "ROLLBACK;", NULL, 0, &err_msg)) {
|
||||
log_error("DB Migration] Unable to ROLLBACK: %s", err_msg);
|
||||
if (err_msg) {
|
||||
sqlite3_free(err_msg);
|
||||
}
|
||||
}
|
||||
|
||||
return FALSE;
|
||||
}
|
||||
|
||||
@@ -1067,9 +1067,12 @@ _handle_groupchat(xmpp_stanza_t* const stanza)
|
||||
char* stanzaid = NULL;
|
||||
xmpp_stanza_t* stanzaidst = xmpp_stanza_get_child_by_name_and_ns(stanza, STANZA_NAME_STANZA_ID, STANZA_NS_STABLE_ID);
|
||||
if (stanzaidst) {
|
||||
stanzaid = (char*)xmpp_stanza_get_attribute(stanzaidst, STANZA_ATTR_ID);
|
||||
if (stanzaid) {
|
||||
message->stanzaid = strdup(stanzaid);
|
||||
const char* by = xmpp_stanza_get_attribute(stanzaidst, "by");
|
||||
if (by && (g_strcmp0(by, from_jid->barejid) == 0)) {
|
||||
stanzaid = (char*)xmpp_stanza_get_attribute(stanzaidst, STANZA_ATTR_ID);
|
||||
if (stanzaid) {
|
||||
message->stanzaid = strdup(stanzaid);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1389,9 +1392,12 @@ _handle_chat(xmpp_stanza_t* const stanza, gboolean is_mam, gboolean is_carbon, c
|
||||
char* stanzaid = NULL;
|
||||
xmpp_stanza_t* stanzaidst = xmpp_stanza_get_child_by_name_and_ns(stanza, STANZA_NAME_STANZA_ID, STANZA_NS_STABLE_ID);
|
||||
if (stanzaidst) {
|
||||
stanzaid = (char*)xmpp_stanza_get_attribute(stanzaidst, STANZA_ATTR_ID);
|
||||
if (stanzaid) {
|
||||
message->stanzaid = strdup(stanzaid);
|
||||
const char* by = xmpp_stanza_get_attribute(stanzaidst, "by");
|
||||
if (by && equals_our_barejid(by)) {
|
||||
stanzaid = (char*)xmpp_stanza_get_attribute(stanzaidst, STANZA_ATTR_ID);
|
||||
if (stanzaid) {
|
||||
message->stanzaid = strdup(stanzaid);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1524,6 +1530,20 @@ _handle_mam(xmpp_stanza_t* const stanza)
|
||||
|
||||
// <result xmlns='urn:xmpp:mam:2' queryid='f27' id='5d398-28273-f7382'>
|
||||
// same as <stanza-id> from XEP-0359 for live messages
|
||||
const char* by = xmpp_stanza_get_attribute(result, "by");
|
||||
const char* from = xmpp_stanza_get_from(stanza);
|
||||
if (by) {
|
||||
if (g_strcmp0(by, from) != 0) {
|
||||
log_warning("MAM result 'by' attribute (%s) does not match 'from' (%s)", by, from);
|
||||
return TRUE;
|
||||
}
|
||||
} else {
|
||||
if (from && !equals_our_barejid(from)) {
|
||||
log_warning("MAM result from %s with no 'by' attribute (expected our own JID)", from);
|
||||
return TRUE;
|
||||
}
|
||||
}
|
||||
|
||||
const char* result_id = xmpp_stanza_get_id(result);
|
||||
|
||||
GDateTime* timestamp = stanza_get_delay_from(forwarded, NULL);
|
||||
|
||||
Reference in New Issue
Block a user