jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Repair existing chat history tables with stable message IDs

master
evgeny 1 week ago
parent
918a3cc061
commit
2f5a49b011
2 changed files with 111 additions and 33 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 25
      src/chat/chat_core.c
  2. 119
      src/chat/db_sync.c

25
src/chat/chat_core.c
Unescape View File

@ -212,19 +212,22 @@ int chat_core_init(struct UTUN_INSTANCE* inst, const char* db_path) {
}
}
/* ensure infrastructure for all existing channels (survives restart) */
/* Сначала закрываем чтение каналов: активный statement блокирует перестройку таблиц истории. */
{
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(cc->db, "SELECT channel_id FROM channels ORDER BY created_at ASC",
-1, &stmt, NULL) == SQLITE_OK) {
int loaded = 0;
while (sqlite3_step(stmt) == SQLITE_ROW) {
const char* ch = (const char*)sqlite3_column_text(stmt, 0);
if (ch && ch[0]) { chat_core_ensure_channel_ready(inst, ch); loaded++; }
}
sqlite3_finalize(stmt);
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "%s: loaded %d channels from DB", CC_ID, loaded);
char** channels = NULL;
int rows = 0, columns = 0, loaded = 0;
int rc = sqlite3_get_table(cc->db, "SELECT channel_id FROM channels ORDER BY created_at ASC",
&channels, &rows, &columns, NULL);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: load channels: %s", CC_ID, sqlite3_errmsg(cc->db));
sqlite3_free_table(channels); chat_core_destroy(inst); return -1;
}
for (int i = 0; i < rows; i++) {
const char* ch = channels[(i + 1) * columns];
if (ch && ch[0]) { chat_core_ensure_channel_ready(inst, ch); loaded++; }
}
sqlite3_free_table(channels);
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "%s: loaded %d channels from DB", CC_ID, loaded);
}
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "%s: initialized, db=%s node_id=0x%016llx",

119
src/chat/db_sync.c
Unescape View File

@ -149,6 +149,99 @@ static int si_prep(struct DB_SYNC_INSTANCE* si, sqlite3_stmt** stmt, const char*
return sqlite3_prepare_v2(SI_DB(si), sql, -1, stmt, NULL);
}
// Одна схема для новых таблиц и однократного преобразования сохранённой истории.
static int si_table_create(struct DB_SYNC_INSTANCE* si, const char* name, int if_not_exists) {
char sql[768];
snprintf(sql, sizeof(sql),
"CREATE TABLE %s\"%s\" ("
" timestamp INTEGER NOT NULL, node_id INTEGER NOT NULL, id INTEGER NOT NULL,"
" chain_hash BLOB NOT NULL, flags INTEGER NOT NULL DEFAULT 0, data BLOB,"
" author_signature BLOB NOT NULL, message_id BLOB NOT NULL UNIQUE CHECK(length(message_id)=32),"
" local_attrs TEXT DEFAULT '', delivered_peers INTEGER NOT NULL DEFAULT 0,"
" delivery_chain TEXT NOT NULL DEFAULT '', PRIMARY KEY(timestamp,author_signature))",
if_not_exists ? "IF NOT EXISTS " : "", name);
if (sqlite3_exec(SI_DB(si), sql, NULL, NULL, NULL) == SQLITE_OK) return 0;
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history schema: create table=%s: %s", name, sqlite3_errmsg(SI_DB(si))); return -1;
}
// Перестройка атомарна: сохраняем подписи, цепочку, локальные атрибуты и доставку без изменений.
static int si_table_init(struct DB_SYNC_INSTANCE* si) {
sqlite3* db = SI_DB(si);
if (si_table_create(si, SI_TBL(si), 1) < 0) return -1;
sqlite3_stmt *rows = NULL, *insert = NULL;
const char* stage = "inspect columns";
int rc = si_prep(si, &rows, "PRAGMA table_info(\"%s\")"), has_id = 0;
if (rc != SQLITE_OK) goto error;
while ((rc = sqlite3_step(rows)) == SQLITE_ROW) {
const char* name = (const char*)sqlite3_column_text(rows, 1);
if (name && !strcmp(name, "message_id")) has_id = 1;
}
if (rc != SQLITE_DONE) goto error;
sqlite3_finalize(rows); rows = NULL;
if (has_id) return 0;
stage = "begin";
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) goto error;
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history schema: adding message_id table=%s group=%016llx",
SI_TBL(si), (unsigned long long)si->group_id);
char replacement[64], sql[768];
snprintf(replacement, sizeof(replacement), "new_%s", SI_TBL(si));
stage = "create replacement";
if (si_table_create(si, replacement, 0) < 0) goto rollback;
stage = "read history";
if (si_prep(si, &rows, "SELECT timestamp,author_signature FROM \"%s\"") != SQLITE_OK) goto rollback;
snprintf(sql, sizeof(sql),
"INSERT INTO \"%s\" SELECT timestamp,node_id,id,chain_hash,flags,data,author_signature,?1,"
"local_attrs,delivered_peers,delivery_chain FROM \"%s\" WHERE timestamp=?2 AND author_signature=?3",
replacement, SI_TBL(si));
stage = "prepare copy";
if (sqlite3_prepare_v2(db, sql, -1, &insert, NULL) != SQLITE_OK) goto rollback;
sqlite3_int64 count = 0;
while ((rc = sqlite3_step(rows)) == SQLITE_ROW) {
sqlite3_int64 ts = sqlite3_column_int64(rows, 0);
const uint8_t* signature = sqlite3_column_blob(rows, 1);
int size = sqlite3_column_bytes(rows, 1);
uint8_t message_id[32];
stage = "compute message_id";
if (!signature || size != DB_SIG_SIZE || chat_message_id(si->group_id, (uint64_t)ts, signature, message_id) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history schema: invalid identity table=%s ts=%lld signature_bytes=%d",
SI_TBL(si), (long long)ts, size); goto rollback;
}
stage = "copy record";
rc = sqlite3_bind_blob(insert, 1, message_id, sizeof(message_id), SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(insert, 2, ts);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(insert, 3, signature, DB_SIG_SIZE, SQLITE_STATIC);
if (rc != SQLITE_OK || sqlite3_step(insert) != SQLITE_DONE) goto rollback;
if (sqlite3_changes(db) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history schema: copy table=%s ts=%lld changed=%d expected=1",
SI_TBL(si), (long long)ts, sqlite3_changes(db)); goto rollback;
}
if (sqlite3_reset(insert) != SQLITE_OK) goto rollback;
count++;
}
stage = "read history";
if (rc != SQLITE_DONE) goto rollback;
sqlite3_finalize(rows); rows = NULL; sqlite3_finalize(insert); insert = NULL;
stage = "replace table";
snprintf(sql, sizeof(sql), "DROP TABLE \"%s\"; ALTER TABLE \"%s\" RENAME TO \"%s\"",
SI_TBL(si), replacement, SI_TBL(si));
if (sqlite3_exec(db, sql, NULL, NULL, NULL) != SQLITE_OK) goto rollback;
stage = "commit";
if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto rollback;
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history schema: message_id ready table=%s records=%lld", SI_TBL(si), (long long)count);
return 0;
rollback:
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history schema: table=%s stage=%s: %s; rolling back",
SI_TBL(si), stage, sqlite3_errmsg(db));
sqlite3_finalize(rows); sqlite3_finalize(insert);
if (sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL) != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history schema: rollback table=%s: %s", SI_TBL(si), sqlite3_errmsg(db));
return -1;
error:
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history schema: table=%s stage=%s: %s", SI_TBL(si), stage, sqlite3_errmsg(db));
sqlite3_finalize(rows); return -1;
}
// ============================================================
// SQLite open/close
// ============================================================
@ -1717,34 +1810,16 @@ struct DB_SYNC_INSTANCE* db_sync_instance_add(struct UTUN_INSTANCE* inst, const
si->group_id = group_id;
snprintf(si->table_name, sizeof(si->table_name), "%s", table_name);
// Create table — PK is (timestamp, node_id), author_signature NOT NULL
if (si_table_init(si) < 0) { db_sync_instance_remove(si); return NULL; }
// Производный TTL-индекс создаём также после перестройки таблицы.
{
char sql[768];
snprintf(sql, sizeof(sql),
"CREATE TABLE IF NOT EXISTS \"%s\" ("
" timestamp INTEGER NOT NULL,"
" node_id INTEGER NOT NULL,"
" id INTEGER NOT NULL,"
" chain_hash BLOB NOT NULL,"
" flags INTEGER NOT NULL DEFAULT 0,"
" data BLOB,"
" author_signature BLOB NOT NULL,"
" message_id BLOB NOT NULL UNIQUE CHECK(length(message_id)=32),"
" local_attrs TEXT DEFAULT '',"
" delivered_peers INTEGER NOT NULL DEFAULT 0,"
" delivery_chain TEXT NOT NULL DEFAULT '',"
" PRIMARY KEY (timestamp, author_signature))",
si->table_name);
int rc = sqlite3_exec(db->db, sql, NULL, NULL, NULL);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "CREATE TABLE %s: %s", si->table_name, sqlite3_errmsg(db->db));
db_sync_instance_remove(si); return NULL;
}
snprintf(sql, sizeof(sql),
"CREATE INDEX IF NOT EXISTS \"idx_%s_ttl\""
" ON \"%s\" (node_id, timestamp)",
si->table_name, si->table_name);
rc = sqlite3_exec(db->db, sql, NULL, NULL, NULL);
int rc = sqlite3_exec(db->db, sql, NULL, NULL, NULL);
if (rc != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "CREATE INDEX %s: %s", si->table_name, sqlite3_errmsg(db->db));
}

Write Preview
Loading…
Cancel
Save