diff --git a/src/chat/chat_core.c b/src/chat/chat_core.c index d62c3783..455b31e6 100644 --- a/src/chat/chat_core.c +++ b/src/chat/chat_core.c @@ -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", diff --git a/src/chat/db_sync.c b/src/chat/db_sync.c index 8a533995..0decca59 100644 --- a/src/chat/db_sync.c +++ b/src/chat/db_sync.c @@ -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)); }