Add atomic DM schema migration and durable author sequence
This commit is contained in:
evgeny committed
1 parent
f21c2582a2
commit
db8f3622d9
2 files changed
+148
No files matched your search
@@ -0,0 +1,132 @@
|
||||
#include "dm_schema.h"
|
||||
#include "../utun_instance.h"
|
||||
#include "../../lib/debug_config.h"
|
||||
#include <sqlite3.h>
|
||||
#include <stdio.h>
|
||||
|
||||
static int schema_error(sqlite3* db, const char* stage) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM schema: %s sql=%s", stage, db ? sqlite3_errmsg(db) : "unavailable");
|
||||
return -1;
|
||||
}
|
||||
|
||||
static int schema_exec(sqlite3* db, const char* sql) {
|
||||
if (sqlite3_exec(db, sql, NULL, NULL, NULL) != SQLITE_OK) return schema_error(db, sql);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Не подменять ошибку чтения схемы отсутствием колонки. */
|
||||
static int schema_has_column(sqlite3* db, const char* table, const char* column) {
|
||||
sqlite3_stmt* st = NULL;
|
||||
int rc = sqlite3_prepare_v2(db, "SELECT 1 FROM pragma_table_info(?) WHERE name=?", -1, &st, NULL);
|
||||
if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 1, table, -1, SQLITE_STATIC);
|
||||
if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 2, column, -1, SQLITE_STATIC);
|
||||
if (rc == SQLITE_OK) rc = sqlite3_step(st);
|
||||
int found = rc == SQLITE_ROW;
|
||||
int final = sqlite3_finalize(st);
|
||||
return (rc == SQLITE_ROW || rc == SQLITE_DONE) && final == SQLITE_OK ? found : schema_error(db, "column lookup");
|
||||
}
|
||||
|
||||
int dm_schema_init(struct UTUN_INSTANCE* inst) {
|
||||
sqlite3* db = inst ? inst->topo_sqlite_db : NULL;
|
||||
if (!db) return schema_error(db, "missing database");
|
||||
if (schema_exec(db, "SAVEPOINT dm_schema")) return -1;
|
||||
const char* conversations =
|
||||
"CREATE TABLE IF NOT EXISTS dm_conversations(conv_id TEXT PRIMARY KEY,peer_node_id INTEGER NOT NULL,"
|
||||
"peer_x25519 BLOB NOT NULL,peer_ed25519 BLOB NOT NULL,peer_name TEXT DEFAULT '',group_id INTEGER DEFAULT 0,"
|
||||
"last_out_seq INTEGER DEFAULT 0,last_in_seq INTEGER DEFAULT 0,created_at INTEGER DEFAULT 0,last_ts INTEGER DEFAULT 0)";
|
||||
const char* messages =
|
||||
"CREATE TABLE IF NOT EXISTS dm_messages(conv_id TEXT NOT NULL,dir INTEGER NOT NULL,seq INTEGER NOT NULL,"
|
||||
"ts INTEGER NOT NULL,author INTEGER NOT NULL,ct TEXT DEFAULT '',data BLOB,sig BLOB,body BLOB,PRIMARY KEY(conv_id,author,seq))";
|
||||
const char* deleted =
|
||||
"CREATE TABLE IF NOT EXISTS dm_deleted_messages(conv_id TEXT NOT NULL,dir INTEGER NOT NULL,seq INTEGER NOT NULL,"
|
||||
"author INTEGER NOT NULL,body_hash BLOB NOT NULL,media_id BLOB,media_receipt BLOB,PRIMARY KEY(conv_id,author,seq))";
|
||||
if (schema_exec(db, conversations)) goto fail;
|
||||
const char* names[] = {"mode", "local_party", "peer_party", "local_key", "closed"};
|
||||
const char* declarations[] = {"INTEGER NOT NULL DEFAULT 0", "BLOB", "BLOB", "BLOB", "INTEGER NOT NULL DEFAULT 0"};
|
||||
for (unsigned i = 0; i < sizeof(names) / sizeof(names[0]); i++) {
|
||||
int found = schema_has_column(db, "dm_conversations", names[i]);
|
||||
if (found < 0) goto fail;
|
||||
if (!found) {
|
||||
char sql[160]; snprintf(sql, sizeof(sql), "ALTER TABLE dm_conversations ADD COLUMN %s %s", names[i], declarations[i]);
|
||||
if (schema_exec(db, sql)) goto fail;
|
||||
}
|
||||
}
|
||||
if (schema_exec(db, messages) || schema_exec(db, deleted)) goto fail;
|
||||
int found = schema_has_column(db, "dm_messages", "body");
|
||||
if (found < 0) goto fail;
|
||||
if (!found) {
|
||||
if (schema_exec(db, "ALTER TABLE dm_messages RENAME TO dm_messages_previous") || schema_exec(db, messages) ||
|
||||
schema_exec(db, "INSERT INTO dm_messages(conv_id,dir,seq,ts,author,ct,data,sig)"
|
||||
" SELECT conv_id,dir,seq,ts,author,ct,data,sig FROM dm_messages_previous") ||
|
||||
schema_exec(db, "DROP TABLE dm_messages_previous")) goto fail;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM schema: retained DEVICE history; message identity now includes author");
|
||||
}
|
||||
found = schema_has_column(db, "dm_deleted_messages", "author");
|
||||
if (found < 0) goto fail;
|
||||
if (!found) {
|
||||
if (schema_exec(db, "ALTER TABLE dm_deleted_messages RENAME TO dm_deleted_previous") || schema_exec(db, deleted)) goto fail;
|
||||
sqlite3_stmt* st = NULL;
|
||||
int rc = sqlite3_prepare_v2(db, "INSERT INTO dm_deleted_messages(conv_id,dir,seq,author,body_hash,media_id,media_receipt)"
|
||||
" SELECT d.conv_id,d.dir,d.seq,CASE WHEN d.dir=1 THEN ? ELSE c.peer_node_id END,d.body_hash,d.media_id,d.media_receipt"
|
||||
" FROM dm_deleted_previous d LEFT JOIN dm_conversations c USING(conv_id)", -1, &st, NULL);
|
||||
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)inst->node_id);
|
||||
if (rc == SQLITE_OK) rc = sqlite3_step(st);
|
||||
int final = sqlite3_finalize(st);
|
||||
if (rc != SQLITE_DONE || final != SQLITE_OK) { schema_error(db, "retain deleted message fingerprints"); goto fail; }
|
||||
if (schema_exec(db, "DROP TABLE dm_deleted_previous")) goto fail;
|
||||
}
|
||||
if (schema_exec(db,
|
||||
"CREATE INDEX IF NOT EXISTS dm_messages_time ON dm_messages(conv_id,ts,author,seq);"
|
||||
"CREATE TABLE IF NOT EXISTS dm_outbox(conv_id TEXT NOT NULL,seq INTEGER NOT NULL,body BLOB NOT NULL,"
|
||||
"retry_at INTEGER NOT NULL DEFAULT 0,PRIMARY KEY(conv_id,seq));"
|
||||
"CREATE TABLE IF NOT EXISTS dm_delivery(conv_id TEXT NOT NULL,author INTEGER NOT NULL,seq INTEGER NOT NULL,"
|
||||
"recipient INTEGER NOT NULL,receipt BLOB,first_attempt INTEGER NOT NULL DEFAULT 0,direct_until INTEGER NOT NULL DEFAULT 0,"
|
||||
"retry_at INTEGER NOT NULL DEFAULT 0,supernode INTEGER NOT NULL DEFAULT 0,PRIMARY KEY(conv_id,author,seq,recipient));"
|
||||
"CREATE TABLE IF NOT EXISTS dm_sequence(node_id INTEGER PRIMARY KEY,value INTEGER NOT NULL);")) goto fail;
|
||||
sqlite3_stmt* st = NULL;
|
||||
int rc = sqlite3_prepare_v2(db, "INSERT INTO dm_sequence(node_id,value) SELECT ?,COALESCE(MAX(last_out_seq),0) FROM dm_conversations"
|
||||
" WHERE 1 ON CONFLICT(node_id) DO UPDATE SET value=MAX(value,excluded.value)", -1, &st, NULL);
|
||||
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)inst->node_id);
|
||||
if (rc == SQLITE_OK) rc = sqlite3_step(st);
|
||||
int final = sqlite3_finalize(st); st = NULL;
|
||||
if (rc != SQLITE_DONE || final != SQLITE_OK) { schema_error(db, "initialize author sequence"); goto fail; }
|
||||
rc = sqlite3_prepare_v2(db, "INSERT INTO dm_delivery(conv_id,author,seq,recipient,retry_at)"
|
||||
" SELECT o.conv_id,?,o.seq,c.peer_node_id,o.retry_at FROM dm_outbox o JOIN dm_conversations c USING(conv_id)"
|
||||
" WHERE c.mode=0 ON CONFLICT(conv_id,author,seq,recipient) DO NOTHING", -1, &st, NULL);
|
||||
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)inst->node_id);
|
||||
if (rc == SQLITE_OK) rc = sqlite3_step(st);
|
||||
final = sqlite3_finalize(st);
|
||||
if (rc != SQLITE_DONE || final != SQLITE_OK) { schema_error(db, "retain DEVICE outbox recipients"); goto fail; }
|
||||
if (schema_exec(db, "RELEASE dm_schema")) goto fail;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM schema: ready; author sequence, frozen parties and per-device delivery");
|
||||
return 0;
|
||||
fail:
|
||||
schema_exec(db, "ROLLBACK TO dm_schema"); schema_exec(db, "RELEASE dm_schema");
|
||||
return -1;
|
||||
}
|
||||
|
||||
int dm_sequence_next(struct UTUN_INSTANCE* inst, uint64_t* seq) {
|
||||
sqlite3* db = inst ? inst->topo_sqlite_db : NULL; sqlite3_stmt* st = NULL;
|
||||
if (!db || !seq) return schema_error(db, "invalid sequence arguments");
|
||||
int rc = sqlite3_prepare_v2(db, "SELECT value FROM dm_sequence WHERE node_id=?", -1, &st, NULL);
|
||||
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)inst->node_id);
|
||||
if (rc == SQLITE_OK) rc = sqlite3_step(st);
|
||||
sqlite3_int64 value = rc == SQLITE_ROW ? sqlite3_column_int64(st, 0) : -1;
|
||||
int final = sqlite3_finalize(st);
|
||||
if (rc != SQLITE_ROW || final != SQLITE_OK || value < 0 || value == INT64_MAX) return schema_error(db, "sequence unavailable/exhausted");
|
||||
*seq = (uint64_t)value + 1;
|
||||
return 0;
|
||||
}
|
||||
|
||||
int dm_sequence_take(struct UTUN_INSTANCE* inst, uint64_t seq) {
|
||||
sqlite3* db = inst ? inst->topo_sqlite_db : NULL; sqlite3_stmt* st = NULL;
|
||||
if (!db || !seq || seq > INT64_MAX || sqlite3_get_autocommit(db)) return schema_error(db, "sequence outside send transaction");
|
||||
int rc = sqlite3_prepare_v2(db, "UPDATE dm_sequence SET value=? WHERE node_id=? AND value=?", -1, &st, NULL);
|
||||
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)seq);
|
||||
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)inst->node_id);
|
||||
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)(seq - 1));
|
||||
if (rc == SQLITE_OK) rc = sqlite3_step(st);
|
||||
int changed = sqlite3_changes(db), final = sqlite3_finalize(st);
|
||||
if (rc != SQLITE_DONE || final != SQLITE_OK || changed != 1) return schema_error(db, "sequence reservation conflict");
|
||||
return 0;
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
/* dm_schema — атомарное создание/обновление локальной схемы личных бесед.
|
||||
* Сохраняет историю, счётчики, точные тела outbox и отпечатки удалённых сообщений.
|
||||
* Изменения выполняются внутри savepoint в uasync-потоке; ошибка откатывает схему. */
|
||||
#ifndef UTUN_DM_SCHEMA_H
|
||||
#define UTUN_DM_SCHEMA_H
|
||||
#include <stdint.h>
|
||||
struct UTUN_INSTANCE;
|
||||
|
||||
/* 0/-1: создать схему и перенести существующие DEVICE-записи без переподписи. */
|
||||
int dm_schema_init(struct UTUN_INSTANCE* inst);
|
||||
/* 0/-1: прочитать следующий seq автора. Резервирование выполняет take в той же
|
||||
* транзакции, что запись сообщения; до commit тело нельзя передавать в сеть. */
|
||||
int dm_sequence_next(struct UTUN_INSTANCE* inst, uint64_t* seq);
|
||||
/* 0/-1: увеличить счётчик с seq-1 до seq. Вызывать внутри транзакции отправки. */
|
||||
int dm_sequence_take(struct UTUN_INSTANCE* inst, uint64_t seq);
|
||||
#endif
|
||||
Reference in new issue
Block a user