1268 lines
71 KiB
C
1268 lines
71 KiB
C
/* db_sync — подписанная история SQLite, модель MESSAGES общего Merkle и адресные PUSH/ACK. */
|
|
#include "db_sync.h"
|
|
#include "merkle_sync.h"
|
|
#include "member_sync.h"
|
|
#include "chat_feedback.h"
|
|
#include "../utun_instance.h"
|
|
#include "../routing_layer/topo_group.h"
|
|
#include "../routing_layer/topo_node_sqlite.h"
|
|
#include "../transport_layer/etcp_api.h"
|
|
#include "../transport_layer/etcp.h"
|
|
#include "../transport_layer/secure_channel.h"
|
|
#include "../../lib/debug_config.h"
|
|
#include "../../lib/mem.h"
|
|
#include "../../lib/u_async.h"
|
|
#include "../../lib/platform_compat.h"
|
|
#ifdef UTUN_HAVE_STANDBY
|
|
#include "standby.h"
|
|
#endif
|
|
#include <openssl/evp.h>
|
|
#include <string.h>
|
|
|
|
#define DB_BATCH_RECORDS 16
|
|
#define DB_WIRE_FIXED (32 + 8 + 8 + 16 + 8 + 8 + 4 + DB_SIG_SIZE)
|
|
#define DB_WIRE_MAX (DB_WIRE_FIXED + DB_SYNC_MESSAGE_MAX)
|
|
#define SI_DB(si) ((si)->db_sync->db)
|
|
#define SI_TBL(si) ((si)->table_name)
|
|
|
|
struct SI_PEER {
|
|
uint64_t node_id;
|
|
uint8_t sync_state, sound_on, paused;
|
|
};
|
|
struct DB_SYNC_INSTANCE {
|
|
struct DB_SYNC* db_sync;
|
|
uint64_t group_id, next_id, last_timestamp_ms;
|
|
char table_name[64], ns[21];
|
|
uint8_t enabled, sync_gated;
|
|
unsigned holds;
|
|
void* ttl_timer;
|
|
struct {
|
|
uint8_t pending, restart, phase, after_valid, waiting_key;
|
|
sqlite3_int64 after;
|
|
void* timer;
|
|
} index;
|
|
struct SI_PEER* peers;
|
|
int peer_count, peer_capacity;
|
|
db_sync_insert_cb on_insert;
|
|
void* on_insert_arg;
|
|
};
|
|
struct db_sync_done_cbk_entry {
|
|
db_sync_done_fn fn;
|
|
void* arg;
|
|
uint8_t removed;
|
|
struct db_sync_done_cbk_entry* next;
|
|
};
|
|
struct db_notice {
|
|
struct db_notice* next;
|
|
struct DB_SYNC* db;
|
|
uint64_t group_id;
|
|
uint8_t initial;
|
|
unsigned count;
|
|
void* work;
|
|
/* Записи и тела лежат непосредственно после этой структуры. */
|
|
};
|
|
struct DB_SYNC {
|
|
struct UTUN_INSTANCE* inst;
|
|
sqlite3* db;
|
|
uint8_t enabled;
|
|
struct DB_SYNC_INSTANCE** instances;
|
|
int instance_count, instance_capacity;
|
|
void* peer_check_timer;
|
|
#ifdef UTUN_HAVE_STANDBY
|
|
void* peer_check_wait; /* Отменяемое ожидание активной фазы Android. */
|
|
#endif
|
|
struct db_sync_done_cbk_entry* done_cbks;
|
|
unsigned dispatch_depth;
|
|
struct db_notice* notices;
|
|
};
|
|
struct db_record {
|
|
sqlite3_int64 rowid;
|
|
uint64_t id, timestamp, author;
|
|
size_t data_len, sig_len;
|
|
uint8_t signature[64], message_id[32];
|
|
struct db_message_revision revision;
|
|
const char* data;
|
|
int cache_valid, updated;
|
|
};
|
|
struct db_checked {
|
|
struct db_record record;
|
|
int check;
|
|
char data[DB_SYNC_MESSAGE_MAX];
|
|
};
|
|
|
|
static void db_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry);
|
|
static void db_sync_on_peer_ready(struct TOPO_GROUP* group, uint64_t peer, int ready, void* arg);
|
|
static void db_sync_peer_check_cb(void* arg);
|
|
static void db_sync_resume_peer_check(struct DB_SYNC* db);
|
|
static void db_index_tick(void* arg);
|
|
static void db_index_request(struct DB_SYNC_INSTANCE* si);
|
|
static int db_start_peer(struct DB_SYNC_INSTANCE* si, uint64_t peer);
|
|
static int db_sync_peer_throttled(struct DB_SYNC* db, struct SI_PEER* peer);
|
|
static int db_message_hash(void*, const char*, uint8_t, uint64_t, EVP_MD_CTX*);
|
|
static const struct merkle_sync_data_ops db_message_ops;
|
|
|
|
static uint64_t db_read64(const uint8_t* p) { uint64_t n; memcpy(&n, p, 8); return be64toh(n); }
|
|
static void db_write64(uint8_t* p, uint64_t n) { n = htobe64(n); memcpy(p, &n, 8); }
|
|
static uint32_t db_read32(const uint8_t* p) { uint32_t n; memcpy(&n, p, 4); return ntohl(n); }
|
|
static void db_write32(uint8_t* p, uint32_t n) { n = htonl(n); memcpy(p, &n, 4); }
|
|
|
|
/* Канонические подписанные байты: домен, канал, постоянный ID, время, автор, nonce, версия, правка, длина и тело. */
|
|
static int db_signed_bytes(uint64_t group, const struct db_record* r, uint8_t* out) {
|
|
static const char domain[] = "utun-chat-record-v2";
|
|
if (!r->data || !r->data_len || r->data_len > DB_SYNC_MESSAGE_MAX || !r->timestamp || r->timestamp > INT64_MAX ||
|
|
!r->author || !r->revision.ver || r->revision.ver > INT64_MAX || r->revision.edited_at > INT64_MAX ||
|
|
(r->revision.ver == 1 ? r->revision.edited_at != 0 : r->revision.edited_at <= r->timestamp)) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history invalid envelope group=%llu ts=%llu ver=%llu edited=%llu bytes=%zu",
|
|
(unsigned long long)group, (unsigned long long)r->timestamp, (unsigned long long)r->revision.ver,
|
|
(unsigned long long)r->revision.edited_at, r->data_len); return -1;
|
|
}
|
|
size_t off = sizeof(domain) - 1; memcpy(out, domain, off); db_write64(out + off, group); off += 8;
|
|
memcpy(out + off, r->message_id, 32); off += 32;
|
|
db_write64(out + off, r->timestamp); off += 8; db_write64(out + off, r->author); off += 8;
|
|
memcpy(out + off, r->revision.nonce, 16); off += 16;
|
|
db_write64(out + off, r->revision.ver); off += 8; db_write64(out + off, r->revision.edited_at); off += 8;
|
|
db_write32(out + off, (uint32_t)r->data_len); off += 4; memcpy(out + off, r->data, r->data_len);
|
|
return (int)(off + r->data_len);
|
|
}
|
|
|
|
int db_sync_sign_message(uint64_t group, uint64_t timestamp, uint64_t author,
|
|
const struct db_message_revision* revision, const char* data, size_t len,
|
|
const uint8_t private_key[32], uint8_t signature[64]) {
|
|
if (!revision || !private_key || !signature) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history invalid signing arguments"); return -1;
|
|
}
|
|
struct db_record r = {.timestamp = timestamp, .author = author, .revision = *revision, .data = data, .data_len = len};
|
|
uint8_t body[DB_WIRE_MAX + 64];
|
|
if (chat_message_id(group, timestamp, author, revision->nonce, r.message_id) < 0) return -1;
|
|
int bytes = db_signed_bytes(group, &r, body);
|
|
if (bytes < 0 || sc_ed25519_sign(private_key, body, (size_t)bytes, signature) != SC_OK) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history signing failed group=%llu ver=%llu",
|
|
(unsigned long long)group, (unsigned long long)revision->ver); return -1;
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
static int db_error(struct DB_SYNC_INSTANCE* si, const char* stage) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history tbl=%s stage=%s: %s", SI_TBL(si), stage, sqlite3_errmsg(SI_DB(si)));
|
|
return -1;
|
|
}
|
|
static void db_rollback(struct DB_SYNC_INSTANCE* si, const char* stage) {
|
|
db_error(si, stage);
|
|
if (sqlite3_exec(SI_DB(si), "ROLLBACK", NULL, NULL, NULL) != SQLITE_OK) db_error(si, "rollback");
|
|
}
|
|
// Форматирует SQL-запрос, подставляя имя таблицы из инстанса (через %s)
|
|
static void si_sql(char* buf, size_t sz, struct DB_SYNC_INSTANCE* si, const char* fmt)
|
|
{
|
|
snprintf(buf, sz, fmt, SI_TBL(si));
|
|
}
|
|
|
|
// Готовит SQLite statement с подстановкой имени таблицы — обёртка над si_sql + sqlite3_prepare_v2
|
|
static int si_prep(struct DB_SYNC_INSTANCE* si, sqlite3_stmt** stmt, const char* fmt)
|
|
{
|
|
char sql[512];
|
|
si_sql(sql, sizeof(sql), si, fmt);
|
|
return sqlite3_prepare_v2(SI_DB(si), sql, -1, stmt, NULL);
|
|
}
|
|
|
|
|
|
static struct DB_SYNC_INSTANCE* db_instance_find(struct DB_SYNC* db, uint64_t group) {
|
|
for (int i = 0; db && i < db->instance_count; i++)
|
|
if (db->instances[i]->enabled && db->instances[i]->group_id == group) return db->instances[i];
|
|
return NULL;
|
|
}
|
|
struct DB_SYNC_INSTANCE* db_sync_instance_find(struct UTUN_INSTANCE* inst, uint64_t group) {
|
|
return db_instance_find(inst ? inst->db_sync : NULL, group);
|
|
}
|
|
uint64_t db_sync_instance_group_id(struct DB_SYNC_INSTANCE* si) { return si ? si->group_id : 0; }
|
|
static void db_release(struct DB_SYNC_INSTANCE* si) {
|
|
if (--si->holds == 0 && !si->enabled) u_free(si);
|
|
}
|
|
static int si_name_valid(const char* name) {
|
|
if (!name || !*name || strlen(name) > 48) return 0;
|
|
for (const char* p = name; *p; p++)
|
|
if (!((*p >= 'a' && *p <= 'z') || (*p >= 'A' && *p <= 'Z') || (*p >= '0' && *p <= '9') || *p == '_')) return 0;
|
|
return 1;
|
|
}
|
|
// Ищет пира в инстансе по node_id
|
|
static struct SI_PEER* si_peer_find(struct DB_SYNC_INSTANCE* si, uint64_t node_id)
|
|
{
|
|
for (int i = 0; i < si->peer_count; i++) {
|
|
if (si->peers[i].node_id == node_id) return &si->peers[i];
|
|
}
|
|
return NULL;
|
|
}
|
|
|
|
// Добавляет пира в инстанс (если уже существует — возвращает существующего)
|
|
static struct SI_PEER* si_peer_add(struct DB_SYNC_INSTANCE* si, uint64_t node_id)
|
|
{
|
|
struct SI_PEER* p = si_peer_find(si, node_id);
|
|
if (p) return p;
|
|
if (si->peer_count >= si->peer_capacity) {
|
|
int nc = si->peer_capacity ? si->peer_capacity * 2 : 8;
|
|
struct SI_PEER* np = u_realloc(si->peers, nc * sizeof(*si->peers));
|
|
if (!np) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history peer allocation failed group=%016llx", si->group_id); return NULL; }
|
|
si->peers = np;
|
|
si->peer_capacity = nc;
|
|
}
|
|
p = &si->peers[si->peer_count++];
|
|
memset(p, 0, sizeof(*p));
|
|
p->node_id = node_id;
|
|
p->sound_on = 1;
|
|
return p;
|
|
}
|
|
|
|
// Получает Ed25519 публичный ключ узла: сначала из локального, затем из topo_node БД, затем из ETCP-соединения
|
|
static int db_get_ed25519_pubkey(struct DB_SYNC* db, uint64_t node_id, uint8_t out[32])
|
|
{
|
|
if (node_id == db->inst->node_id) { memcpy(out, db->inst->my_ed25519_pubkey, 32); return 0; }
|
|
/* из in-memory registry (BGP/topo, self-sig верифицирован) — первичный источник */
|
|
if (db->inst->topo_groups) {
|
|
struct TOPO_NODE* ni = topo_node_registry_find(db->inst->topo_groups, node_id);
|
|
if (ni) {
|
|
uint64_t chk; memcpy(&chk, ni->ed25519_public_key, 8);
|
|
if (chk != 0) { memcpy(out, ni->ed25519_public_key, 32); return 0; }
|
|
}
|
|
}
|
|
if (db->inst->topo_sqlite_db) {
|
|
int rc = topo_node_sqlite_get_ed25519_pubkey(db->inst->topo_sqlite_db, node_id, out);
|
|
if (rc == 0) return 0;
|
|
if (rc == TOPO_SQLITE_LOOKUP_ERROR) return -1;
|
|
}
|
|
struct ETCP_CONN* conn = instance_find_conn(db->inst, node_id);
|
|
if (conn) {
|
|
memcpy(out, conn->peer_ed25519_pubkey, 32);
|
|
uint64_t chk; memcpy(&chk, out, 8);
|
|
if (chk != 0) return 0;
|
|
}
|
|
return 1; /* Ключ ещё не получен; это не доказательство порчи. */
|
|
}
|
|
|
|
enum db_record_check { DB_CHECK_ERROR = -1, DB_CHECK_VALID, DB_CHECK_BAD_SIGNATURE, DB_CHECK_BAD_FORMAT, DB_CHECK_DEFERRED };
|
|
|
|
/* Проверяем конкретные байты. Только BAD_* доказывает их непригодность. */
|
|
static enum db_record_check db_verify_author_sig(struct DB_SYNC_INSTANCE* si, const struct db_record* r)
|
|
{
|
|
uint8_t msg[DB_WIRE_MAX + 64];
|
|
int bytes = db_signed_bytes(si->group_id, r, msg);
|
|
if (r->sig_len != DB_SIG_SIZE || bytes < 0) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history malformed record tbl=%s ts=%llu body=%zu signature=%zu",
|
|
SI_TBL(si), (unsigned long long)r->timestamp, r->data_len, r->sig_len); return DB_CHECK_BAD_FORMAT;
|
|
}
|
|
uint8_t pubkey[32];
|
|
int key_rc = db_get_ed25519_pubkey(si->db_sync, r->author, pubkey);
|
|
if (key_rc != 0) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history verification unavailable tbl=%s author=%016llx key_status=%d; preserving data",
|
|
SI_TBL(si), (unsigned long long)r->author, key_rc);
|
|
return key_rc < 0 ? DB_CHECK_ERROR : DB_CHECK_DEFERRED;
|
|
}
|
|
int rc = sc_ed25519_verify(pubkey, msg, (size_t)bytes, r->signature);
|
|
if (rc != SC_OK) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history signature check tbl=%s ts=%llu author=%016llx rc=%d result=%s",
|
|
SI_TBL(si), (unsigned long long)r->timestamp, (unsigned long long)r->author, rc,
|
|
rc == SC_ERR_AUTH_FAILED ? "mismatch" : "operational failure; preserving data");
|
|
return rc == SC_ERR_AUTH_FAILED ? DB_CHECK_BAD_SIGNATURE : DB_CHECK_ERROR;
|
|
}
|
|
return DB_CHECK_VALID;
|
|
}
|
|
|
|
/* Вызывается после завершения SELECT и только внутри собственной транзакции. */
|
|
static int db_delete_bad_row(struct DB_SYNC_INSTANCE* si, sqlite3_int64 rowid) {
|
|
sqlite3_stmt* stmt = NULL;
|
|
int rc = si_prep(si, &stmt, "DELETE FROM \"%s\" WHERE rowid=?");
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(stmt, 1, rowid);
|
|
if (rc == SQLITE_OK) rc = sqlite3_step(stmt);
|
|
int changed = sqlite3_changes(SI_DB(si)), final_rc = sqlite3_finalize(stmt);
|
|
if (rc == SQLITE_DONE && final_rc == SQLITE_OK && changed == 1) return 0;
|
|
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history delete tbl=%s rowid=%lld rc=%d finalize=%d changed=%d: %s",
|
|
SI_TBL(si), (long long)rowid, rc, final_rc, changed, sqlite3_errmsg(SI_DB(si))); return -1;
|
|
}
|
|
|
|
// Формат v2: одна строка на постоянный ID, подпись и версия заменяются атомарно.
|
|
static int si_table_create(struct DB_SYNC_INSTANCE* si, const char* name, int if_not_exists) {
|
|
char sql[1024];
|
|
snprintf(sql, sizeof(sql),
|
|
"CREATE TABLE %s\"%s\" ("
|
|
" timestamp INTEGER NOT NULL, node_id INTEGER NOT NULL, id INTEGER 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),"
|
|
" bucket INTEGER NOT NULL, local_attrs TEXT DEFAULT '', delivered_peers INTEGER NOT NULL DEFAULT 0,"
|
|
" delivery_chain TEXT NOT NULL DEFAULT '', nonce BLOB NOT NULL CHECK(length(nonce)=16),"
|
|
" ver INTEGER NOT NULL CHECK(ver>0), edited_at INTEGER NOT NULL CHECK(edited_at>=0))",
|
|
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;
|
|
}
|
|
|
|
/* Dev-формат без переноса: удаляем только старую групповую историю и зависимые feedback.
|
|
* PM, участники, ключи и прочие таблицы общей БД сохраняются. DDL и очистка индекса атомарны. */
|
|
static int si_table_init(struct DB_SYNC_INSTANCE* si) {
|
|
sqlite3* db = SI_DB(si); sqlite3_stmt* st = NULL;
|
|
if (si_table_create(si, SI_TBL(si), 1) < 0) return -1;
|
|
int rc = si_prep(si, &st, "PRAGMA table_info(\"%s\")"), versioned = 0;
|
|
if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
|
|
const char* name = (const char*)sqlite3_column_text(st, 1);
|
|
if (name && !strcmp(name, "nonce")) versioned = 1;
|
|
}
|
|
int final_rc = sqlite3_finalize(st); st = NULL;
|
|
if (rc != SQLITE_DONE || final_rc != SQLITE_OK) return db_error(si, "inspect history format");
|
|
if (versioned) return 0;
|
|
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) return db_error(si, "format reset begin");
|
|
char sql[512];
|
|
snprintf(sql, sizeof(sql), "DROP TABLE \"%s\"", SI_TBL(si));
|
|
if (sqlite3_exec(db, sql, NULL, NULL, NULL) != SQLITE_OK || si_table_create(si, SI_TBL(si), 0) < 0) goto rollback;
|
|
/* В тестах произвольные таблицы не имеют feedback; в чате он подготовлен до регистрации истории. */
|
|
snprintf(sql, sizeof(sql), "SELECT 1 FROM sqlite_master WHERE type='table' AND name='feedback_%s'", si->ns);
|
|
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) goto rollback;
|
|
rc = sqlite3_step(st); final_rc = sqlite3_finalize(st); st = NULL;
|
|
if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || final_rc != SQLITE_OK) goto rollback;
|
|
if (rc == SQLITE_ROW) {
|
|
snprintf(sql, sizeof(sql), "DELETE FROM \"feedback_%s\"", si->ns);
|
|
if (sqlite3_exec(db, sql, NULL, NULL, NULL) != SQLITE_OK) goto rollback;
|
|
}
|
|
char messages[25], feedback[25];
|
|
snprintf(messages, sizeof(messages), "%u:%s", MT_DATA_MESSAGES, si->ns);
|
|
snprintf(feedback, sizeof(feedback), "%u:%s", MT_DATA_FEEDBACK, si->ns);
|
|
const char* dependencies[] = {"merkle_tree_hash", "attachment_keys", "attachment_key_requests", "attachment_key_outbox", "channels", "ui_state"};
|
|
const char* cleanup[] = {
|
|
"DELETE FROM merkle_tree_hash WHERE namespace IN (?1,?2)",
|
|
"DELETE FROM attachment_keys WHERE channel=?1", "DELETE FROM attachment_key_requests WHERE channel=?1",
|
|
"DELETE FROM attachment_key_outbox WHERE channel=?1",
|
|
"UPDATE channels SET last_msg_at=0,last_read_msg_id=0,last_pos_msg_id=0 WHERE channel_id=?1",
|
|
"DELETE FROM ui_state WHERE key IN ('ch_'||?1||'_last_read','ch_'||?1||'_last_pos')"
|
|
};
|
|
for (unsigned i = 0; i < sizeof(dependencies) / sizeof(dependencies[0]); i++) {
|
|
if (sqlite3_prepare_v2(db, "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", -1, &st, NULL) != SQLITE_OK) goto rollback;
|
|
rc = sqlite3_bind_text(st, 1, dependencies[i], -1, SQLITE_STATIC);
|
|
if (rc == SQLITE_OK) rc = sqlite3_step(st);
|
|
final_rc = sqlite3_finalize(st); st = NULL;
|
|
if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || final_rc != SQLITE_OK) goto rollback;
|
|
if (rc == SQLITE_DONE) continue;
|
|
if (sqlite3_prepare_v2(db, cleanup[i], -1, &st, NULL) != SQLITE_OK) goto rollback;
|
|
rc = sqlite3_bind_text(st, 1, i ? si->ns : messages, -1, SQLITE_STATIC);
|
|
if (rc == SQLITE_OK && !i) rc = sqlite3_bind_text(st, 2, feedback, -1, SQLITE_STATIC);
|
|
if (rc == SQLITE_OK) rc = sqlite3_step(st);
|
|
final_rc = sqlite3_finalize(st); st = NULL;
|
|
if (rc != SQLITE_DONE || final_rc != SQLITE_OK) goto rollback;
|
|
}
|
|
if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto rollback;
|
|
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history format v2: reset old group history and feedback tbl=%s group=%s",
|
|
SI_TBL(si), si->ns);
|
|
return 0;
|
|
rollback:
|
|
sqlite3_finalize(st); db_rollback(si, "history format reset"); return -1;
|
|
}
|
|
|
|
#define DB_COLUMNS "rowid,id,timestamp,node_id,data,author_signature,message_id,bucket,nonce,ver,edited_at"
|
|
|
|
/* Чтение конкретной строки. cache_valid — производные поля, их можно восстановить без удаления сообщения. */
|
|
static enum db_record_check db_read_record(struct DB_SYNC_INSTANCE* si, sqlite3_stmt* st, struct db_record* r) {
|
|
memset(r, 0, sizeof(*r));
|
|
r->rowid = sqlite3_column_int64(st, 0); r->id = (uint64_t)sqlite3_column_int64(st, 1);
|
|
r->timestamp = (uint64_t)sqlite3_column_int64(st, 2); r->author = (uint64_t)sqlite3_column_int64(st, 3);
|
|
r->data = sqlite3_column_blob(st, 4);
|
|
if (!r->data && sqlite3_errcode(SI_DB(si)) == SQLITE_NOMEM) return DB_CHECK_ERROR;
|
|
r->data_len = (size_t)sqlite3_column_bytes(st, 4);
|
|
const uint8_t* sig = sqlite3_column_blob(st, 5);
|
|
if (!sig && sqlite3_errcode(SI_DB(si)) == SQLITE_NOMEM) return DB_CHECK_ERROR;
|
|
r->sig_len = (size_t)sqlite3_column_bytes(st, 5);
|
|
if (!r->data || !r->data_len || r->data_len > DB_SYNC_MESSAGE_MAX || !sig || r->sig_len != 64) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history malformed stored row tbl=%s rowid=%lld body=%zu signature=%zu",
|
|
SI_TBL(si), (long long)r->rowid, r->data_len, r->sig_len); return DB_CHECK_BAD_FORMAT;
|
|
}
|
|
memcpy(r->signature, sig, 64);
|
|
const void* nonce = sqlite3_column_blob(st, 8);
|
|
if (!nonce && sqlite3_errcode(SI_DB(si)) == SQLITE_NOMEM) return DB_CHECK_ERROR;
|
|
if (!nonce || sqlite3_column_bytes(st, 8) != 16) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history malformed nonce tbl=%s rowid=%lld", SI_TBL(si), (long long)r->rowid);
|
|
return DB_CHECK_BAD_FORMAT;
|
|
}
|
|
memcpy(r->revision.nonce, nonce, 16);
|
|
r->revision.ver = (uint64_t)sqlite3_column_int64(st, 9);
|
|
r->revision.edited_at = (uint64_t)sqlite3_column_int64(st, 10);
|
|
if (chat_message_id(si->group_id, r->timestamp, r->author, r->revision.nonce, r->message_id) < 0) return DB_CHECK_ERROR;
|
|
const uint8_t* key = sqlite3_column_blob(st, 6);
|
|
if (!key && sqlite3_errcode(SI_DB(si)) == SQLITE_NOMEM) return DB_CHECK_ERROR;
|
|
r->cache_valid = key && sqlite3_column_bytes(st, 6) == 32 && !memcmp(key, r->message_id, 32) &&
|
|
(uint64_t)sqlite3_column_int64(st, 7) == (db_read64(r->message_id) >> 39);
|
|
return DB_CHECK_VALID;
|
|
}
|
|
|
|
static int db_encode_record(const struct db_record* r, uint8_t* buf) {
|
|
memcpy(buf, r->message_id, 32); db_write64(buf + 32, r->timestamp); db_write64(buf + 40, r->author);
|
|
memcpy(buf + 48, r->revision.nonce, 16); db_write64(buf + 64, r->revision.ver); db_write64(buf + 72, r->revision.edited_at);
|
|
db_write32(buf + 80, (uint32_t)r->data_len); memcpy(buf + 84, r->data, r->data_len);
|
|
memcpy(buf + 84 + r->data_len, r->signature, 64);
|
|
return (int)(DB_WIRE_FIXED + r->data_len);
|
|
}
|
|
static int db_parse_record(struct DB_SYNC_INSTANCE* si, const uint8_t** ptr, const uint8_t* end, struct db_record* r) {
|
|
if ((size_t)(end - *ptr) < DB_WIRE_FIXED) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history truncated record"); return -1; }
|
|
const uint8_t* p = *ptr;
|
|
memset(r, 0, sizeof(*r));
|
|
r->timestamp = db_read64(p + 32); r->author = db_read64(p + 40); r->data_len = db_read32(p + 80);
|
|
memcpy(r->revision.nonce, p + 48, 16); r->revision.ver = db_read64(p + 64); r->revision.edited_at = db_read64(p + 72);
|
|
if (!r->data_len || r->data_len > DB_SYNC_MESSAGE_MAX || r->data_len > (size_t)(end - p) - DB_WIRE_FIXED) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history invalid wire body=%zu", r->data_len); return -1;
|
|
}
|
|
r->data = (const char*)p + 84; r->sig_len = 64; memcpy(r->signature, p + 84 + r->data_len, 64);
|
|
if (chat_message_id(si->group_id, r->timestamp, r->author, r->revision.nonce, r->message_id) < 0) return -1;
|
|
if (memcmp(r->message_id, p, 32)) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history wire identity mismatch"); return -1; }
|
|
*ptr = p + DB_WIRE_FIXED + r->data_len;
|
|
return 0;
|
|
}
|
|
|
|
static int db_index_clear(struct DB_SYNC_INSTANCE* si) {
|
|
char ns[25]; snprintf(ns, sizeof(ns), "%u:%s", MT_DATA_MESSAGES, si->ns);
|
|
sqlite3_stmt* st = NULL;
|
|
int rc = sqlite3_prepare_v2(SI_DB(si), "DELETE FROM merkle_tree_hash WHERE namespace=?", -1, &st, NULL);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 1, ns, -1, SQLITE_STATIC);
|
|
if (rc == SQLITE_OK) rc = sqlite3_step(st);
|
|
int end_rc = sqlite3_finalize(st);
|
|
return rc == SQLITE_DONE && end_rc == SQLITE_OK ? 0 : db_error(si, "clear message index");
|
|
}
|
|
|
|
static int db_update_path(struct DB_SYNC_INSTANCE* si, const uint8_t key[32]) {
|
|
if (si->index.pending || !merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES)) return 0;
|
|
return merkle_sync_recompute_path(si->db_sync->inst, MT_DATA_MESSAGES, si->ns, db_read64(key)) < 0 ? -1 : 0;
|
|
}
|
|
|
|
static void db_committed(struct DB_SYNC_INSTANCE* si, int changed) {
|
|
if (si->index.pending || (changed & 2)) db_index_request(si);
|
|
else merkle_sync_changed(si->db_sync->inst, MT_DATA_MESSAGES, si->ns);
|
|
}
|
|
|
|
static int db_same_record(const struct db_record* a, const struct db_record* b) {
|
|
return a->timestamp == b->timestamp && a->author == b->author && a->data_len == b->data_len &&
|
|
a->revision.ver == b->revision.ver && a->revision.edited_at == b->revision.edited_at &&
|
|
!memcmp(a->revision.nonce, b->revision.nonce, 16) &&
|
|
a->sig_len == b->sig_len && !memcmp(a->signature, b->signature, 64) && !memcmp(a->data, b->data, a->data_len);
|
|
}
|
|
|
|
/* Внешняя транзакция обязательна. Входящая подпись уже проверена до захвата write-lock. */
|
|
static int db_put_record(struct DB_SYNC_INSTANCE* si, struct db_record* r, enum db_record_check incoming,
|
|
const char* attrs, int* changed) {
|
|
sqlite3_stmt* st = NULL;
|
|
if (si_prep(si, &st, "SELECT " DB_COLUMNS " FROM \"%s\" WHERE message_id=?") != SQLITE_OK)
|
|
return db_error(si, "read duplicate");
|
|
int rc = sqlite3_bind_blob(st, 1, r->message_id, 32, SQLITE_STATIC);
|
|
if (rc == SQLITE_OK) rc = sqlite3_step(st);
|
|
struct db_record stored;
|
|
int exists = rc == SQLITE_ROW, remove = 0;
|
|
if (exists) {
|
|
enum db_record_check check = db_read_record(si, st, &stored);
|
|
if (check == DB_CHECK_VALID && !(incoming == DB_CHECK_VALID && db_same_record(&stored, r)))
|
|
check = db_verify_author_sig(si, &stored);
|
|
if (check == DB_CHECK_ERROR || check == DB_CHECK_DEFERRED) { sqlite3_finalize(st); return -1; }
|
|
remove = check == DB_CHECK_BAD_FORMAT || check == DB_CHECK_BAD_SIGNATURE;
|
|
if (!remove && !stored.cache_valid) {
|
|
sqlite3_finalize(st); db_index_request(si); return db_error(si, "derived identity needs rebuild");
|
|
}
|
|
} else if (rc != SQLITE_DONE) { sqlite3_finalize(st); return db_error(si, "duplicate step"); }
|
|
int same = exists && !remove && db_same_record(&stored, r);
|
|
if (sqlite3_finalize(st) != SQLITE_OK) return db_error(si, "duplicate finalize");
|
|
if (remove) {
|
|
if (db_delete_bad_row(si, stored.rowid) < 0) return -1;
|
|
if (merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES) && db_index_clear(si) < 0) return -1;
|
|
*changed |= 3;
|
|
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history staged deletion of proven corrupt duplicate tbl=%s rowid=%lld",
|
|
SI_TBL(si), (long long)stored.rowid);
|
|
}
|
|
if (incoming != DB_CHECK_VALID) return -2;
|
|
if (exists && !remove) {
|
|
if (r->revision.ver < stored.revision.ver || same) {
|
|
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history duplicate/stale tbl=%s id=%02x%02x%02x%02x incoming=%llu stored=%llu",
|
|
SI_TBL(si), r->message_id[0], r->message_id[1], r->message_id[2], r->message_id[3],
|
|
(unsigned long long)r->revision.ver, (unsigned long long)stored.revision.ver); return 1;
|
|
}
|
|
if (r->revision.ver == stored.revision.ver) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history version conflict tbl=%s id=%02x%02x%02x%02x ver=%llu",
|
|
SI_TBL(si), r->message_id[0], r->message_id[1], r->message_id[2], r->message_id[3],
|
|
(unsigned long long)r->revision.ver); return -2;
|
|
}
|
|
r->updated = 1; r->id = stored.id;
|
|
if (si_prep(si, &st, "UPDATE \"%s\" SET data=?,author_signature=?,ver=?,edited_at=?,"
|
|
"flags=flags & ~1,delivered_peers=0,delivery_chain='' WHERE message_id=?") != SQLITE_OK) return db_error(si, "update prepare");
|
|
rc = sqlite3_bind_blob(st, 1, r->data, (int)r->data_len, SQLITE_STATIC);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 2, r->signature, 64, SQLITE_STATIC);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)r->revision.ver);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 4, (sqlite3_int64)r->revision.edited_at);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 5, r->message_id, 32, SQLITE_STATIC);
|
|
if (rc == SQLITE_OK) rc = sqlite3_step(st);
|
|
int end_rc = sqlite3_finalize(st);
|
|
if (rc != SQLITE_DONE || end_rc != SQLITE_OK) return db_error(si, "update step");
|
|
if (db_update_path(si, r->message_id) < 0) return -1;
|
|
*changed |= 1;
|
|
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history updated tbl=%s id=%02x%02x%02x%02x ver=%llu->%llu edited=%llu",
|
|
SI_TBL(si), r->message_id[0], r->message_id[1], r->message_id[2], r->message_id[3],
|
|
(unsigned long long)stored.revision.ver, (unsigned long long)r->revision.ver,
|
|
(unsigned long long)r->revision.edited_at); return 0;
|
|
}
|
|
if (si_prep(si, &st, "INSERT INTO \"%s\" (timestamp,node_id,id,flags,data,author_signature,message_id,bucket,local_attrs,nonce,ver,edited_at)"
|
|
" VALUES(?,?,?,0,?,?,?,?,?,?,?,?)") != SQLITE_OK) return db_error(si, "insert prepare");
|
|
rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)r->timestamp);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)r->author);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)r->id);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 4, r->data, (int)r->data_len, SQLITE_STATIC);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 5, r->signature, 64, SQLITE_STATIC);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 6, r->message_id, 32, SQLITE_STATIC);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 7, (sqlite3_int64)(db_read64(r->message_id) >> 39));
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 8, attrs ? attrs : "", -1, SQLITE_STATIC);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 9, r->revision.nonce, 16, SQLITE_STATIC);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 10, (sqlite3_int64)r->revision.ver);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 11, (sqlite3_int64)r->revision.edited_at);
|
|
if (rc == SQLITE_OK) rc = sqlite3_step(st);
|
|
int final_rc = sqlite3_finalize(st);
|
|
if (rc != SQLITE_DONE || final_rc != SQLITE_OK) return db_error(si, "insert step");
|
|
if (!(*changed & 2) && db_update_path(si, r->message_id) < 0) return -1;
|
|
*changed |= 1; return 0;
|
|
}
|
|
|
|
/* Предварительная проверка идёт без write-lock; внутри короткой транзакции сверяем прочитанные байты. */
|
|
static int db_index_batch(struct DB_SYNC_INSTANCE* si) {
|
|
sqlite3* db = SI_DB(si);
|
|
sqlite3_stmt* st = NULL;
|
|
if (!merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES)) return 1;
|
|
if (si->index.restart) {
|
|
merkle_sync_cancel_dataset_ns(si->db_sync->inst, MT_DATA_MESSAGES, si->ns);
|
|
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) return db_error(si, "index begin");
|
|
if (db_index_clear(si) < 0 || sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) {
|
|
db_rollback(si, "index reset"); return -1;
|
|
}
|
|
si->index.restart = 0; si->index.phase = 0; si->index.after_valid = 0; si->index.after = 0;
|
|
}
|
|
if (si->index.phase == 0) {
|
|
struct db_checked* rows = u_calloc(DB_BATCH_RECORDS, sizeof(*rows));
|
|
if (!rows) return db_error(si, "preflight allocation");
|
|
const char* query = si->index.after_valid
|
|
? "SELECT " DB_COLUMNS " FROM \"%s\" WHERE rowid>? ORDER BY rowid LIMIT 16"
|
|
: "SELECT " DB_COLUMNS " FROM \"%s\" ORDER BY rowid LIMIT 16";
|
|
int rc = si_prep(si, &st, query), count = 0, removed = 0;
|
|
if (rc == SQLITE_OK && si->index.after_valid) rc = sqlite3_bind_int64(st, 1, si->index.after);
|
|
if (rc != SQLITE_OK) goto verify_error;
|
|
while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
|
|
struct db_checked* item = &rows[count];
|
|
item->check = db_read_record(si, st, &item->record);
|
|
if (item->check == DB_CHECK_VALID) {
|
|
memcpy(item->data, item->record.data, item->record.data_len); item->record.data = item->data;
|
|
item->check = db_verify_author_sig(si, &item->record);
|
|
}
|
|
if (item->check == DB_CHECK_ERROR || item->check == DB_CHECK_DEFERRED) {
|
|
si->index.waiting_key = item->check == DB_CHECK_DEFERRED;
|
|
goto verify_error;
|
|
}
|
|
count++;
|
|
}
|
|
int final_rc = sqlite3_finalize(st); st = NULL;
|
|
if (rc != SQLITE_DONE || final_rc != SQLITE_OK) goto verify_error;
|
|
st = NULL;
|
|
if (!count) {
|
|
si->index.phase = 1; si->index.after_valid = 0; si->index.after = -1;
|
|
u_free(rows); return 0;
|
|
}
|
|
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) goto verify_error;
|
|
for (int i = 0; i < count; i++) {
|
|
struct db_checked* item = &rows[i]; struct db_record current;
|
|
if (si_prep(si, &st, "SELECT " DB_COLUMNS " FROM \"%s\" WHERE rowid=?") != SQLITE_OK) goto verify_rollback;
|
|
if (sqlite3_bind_int64(st, 1, item->record.rowid) != SQLITE_OK || sqlite3_step(st) != SQLITE_ROW) goto verify_rollback;
|
|
enum db_record_check now = db_read_record(si, st, ¤t);
|
|
int matches = now == DB_CHECK_BAD_FORMAT && item->check == DB_CHECK_BAD_FORMAT;
|
|
if (now == DB_CHECK_VALID && item->check != DB_CHECK_BAD_FORMAT) matches = db_same_record(&item->record, ¤t);
|
|
if (sqlite3_finalize(st) != SQLITE_OK) { st = NULL; goto verify_rollback; }
|
|
st = NULL;
|
|
if (!matches) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history preflight source changed tbl=%s; retry", SI_TBL(si)); goto verify_rollback;
|
|
}
|
|
if (item->check == DB_CHECK_BAD_FORMAT || item->check == DB_CHECK_BAD_SIGNATURE) {
|
|
if (db_delete_bad_row(si, item->record.rowid) < 0) goto verify_rollback;
|
|
removed++;
|
|
} else {
|
|
if (si_prep(si, &st, "UPDATE \"%s\" SET message_id=?,bucket=? WHERE rowid=?") != SQLITE_OK) goto verify_rollback;
|
|
if (sqlite3_bind_blob(st, 1, item->record.message_id, 32, SQLITE_STATIC) != SQLITE_OK ||
|
|
sqlite3_bind_int64(st, 2, (sqlite3_int64)(db_read64(item->record.message_id) >> 39)) != SQLITE_OK ||
|
|
sqlite3_bind_int64(st, 3, item->record.rowid) != SQLITE_OK || sqlite3_step(st) != SQLITE_DONE) goto verify_rollback;
|
|
if (sqlite3_finalize(st) != SQLITE_OK) { st = NULL; goto verify_rollback; }
|
|
st = NULL;
|
|
}
|
|
}
|
|
if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto verify_rollback;
|
|
si->index.after = rows[count - 1].record.rowid; si->index.after_valid = 1; si->index.waiting_key = 0;
|
|
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history preflight committed tbl=%s checked=%d removed=%d after=%lld",
|
|
SI_TBL(si), count, removed, (long long)si->index.after);
|
|
if (removed) DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history removed %d proven invalid stored records tbl=%s", removed, SI_TBL(si));
|
|
u_free(rows); return 0;
|
|
verify_rollback:
|
|
sqlite3_finalize(st); st = NULL; db_rollback(si, "preflight commit");
|
|
verify_error:
|
|
sqlite3_finalize(st); u_free(rows); return db_error(si, si->index.waiting_key ? "preflight waiting for author key" : "preflight failed");
|
|
}
|
|
|
|
/* Все сохранённые подписи проверены; в write-транзакции остаётся только хеширование ограниченной пачки листьев. */
|
|
sqlite3_int64 buckets[DB_BATCH_RECORDS]; int count = 0, rc;
|
|
if (si_prep(si, &st, "SELECT DISTINCT bucket FROM \"%s\" WHERE bucket>? ORDER BY bucket LIMIT 16") != SQLITE_OK)
|
|
return db_error(si, "index buckets");
|
|
if (sqlite3_bind_int64(st, 1, si->index.after) != SQLITE_OK) { sqlite3_finalize(st); return db_error(si, "bucket bind"); }
|
|
while ((rc = sqlite3_step(st)) == SQLITE_ROW) buckets[count++] = sqlite3_column_int64(st, 0);
|
|
int end_rc = sqlite3_finalize(st);
|
|
if (rc != SQLITE_DONE || end_rc != SQLITE_OK) return db_error(si, "bucket step");
|
|
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) return db_error(si, "hash begin");
|
|
for (int i = 0; i < count; i++)
|
|
if (merkle_sync_recompute_path(si->db_sync->inst, MT_DATA_MESSAGES, si->ns, (uint64_t)buckets[i] << 39) < 0) {
|
|
db_rollback(si, "hash batch"); return -1;
|
|
}
|
|
if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) { db_rollback(si, "hash commit"); return -1; }
|
|
if (count) si->index.after = buckets[count - 1];
|
|
else {
|
|
si->index.pending = 0;
|
|
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history verified index ready tbl=%s group=%s", SI_TBL(si), si->ns);
|
|
merkle_sync_changed(si->db_sync->inst, MT_DATA_MESSAGES, si->ns);
|
|
for (int i = 0; i < si->peer_count; i++) db_start_peer(si, si->peers[i].node_id);
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
static void db_index_tick(void* arg) {
|
|
struct DB_SYNC_INSTANCE* si = arg; si->index.timer = NULL;
|
|
if (!si->enabled || !si->index.pending) return;
|
|
int rc = db_index_batch(si);
|
|
if (si->index.pending && merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES)) {
|
|
si->index.timer = uasync_set_timeout(si->db_sync->inst->ua, rc ? 50000 : 1, si, db_index_tick, "history_preflight");
|
|
if (!si->index.timer) { db_error(si, "preflight timer"); db_sync_resume_peer_check(si->db_sync); }
|
|
}
|
|
}
|
|
static void db_index_request(struct DB_SYNC_INSTANCE* si) {
|
|
si->index.pending = 1; si->index.restart = 1; si->index.waiting_key = 0;
|
|
if (!si->index.timer && merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES))
|
|
si->index.timer = uasync_set_timeout(si->db_sync->inst->ua, 0, si, db_index_tick, "history_preflight");
|
|
db_sync_resume_peer_check(si->db_sync);
|
|
}
|
|
|
|
/* Чтение листа — одинаковый канонический отбор для корня и передачи. Проверка авторства уже выполнена preflight/put. */
|
|
static int db_message_hash(void* ctx, const char* ns, uint8_t level, uint64_t prefix, EVP_MD_CTX* hash) {
|
|
struct DB_SYNC_INSTANCE* si = db_sync_instance_find(ctx, strtoull(ns, NULL, 10));
|
|
if (!si || level != MT_MAX_LEVEL) return -1;
|
|
sqlite3_stmt* st = NULL;
|
|
if (si_prep(si, &st, "SELECT " DB_COLUMNS " FROM \"%s\" WHERE bucket=? ORDER BY message_id") != SQLITE_OK)
|
|
return db_error(si, "leaf prepare");
|
|
int rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)(prefix >> 39)), count = 0, bad = 0;
|
|
if (rc != SQLITE_OK) { sqlite3_finalize(st); return db_error(si, "leaf bind"); }
|
|
while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
|
|
struct db_record r; uint8_t wire[DB_WIRE_MAX];
|
|
enum db_record_check check = db_read_record(si, st, &r);
|
|
if (check != DB_CHECK_VALID || !r.cache_valid) { bad = check != DB_CHECK_ERROR; rc = SQLITE_ERROR; break; }
|
|
int bytes = db_encode_record(&r, wire);
|
|
if (EVP_DigestUpdate(hash, wire, (size_t)bytes) != 1) { rc = SQLITE_ERROR; break; }
|
|
count++;
|
|
}
|
|
int end_rc = sqlite3_finalize(st);
|
|
if (bad) db_index_request(si);
|
|
if (rc != SQLITE_DONE || end_rc != SQLITE_OK) return db_error(si, "leaf read/hash");
|
|
return count;
|
|
}
|
|
|
|
static int db_message_page(void* ctx, const char* ns, uint64_t prefix, int av, const merkle_key_t* after,
|
|
uint8_t* buf, size_t* len, merkle_key_t* next, int* more) {
|
|
struct DB_SYNC_INSTANCE* si = db_sync_instance_find(ctx, strtoull(ns, NULL, 10));
|
|
if (!si || si->index.pending) return -1;
|
|
sqlite3_stmt* st = NULL;
|
|
const char* query = av
|
|
? "SELECT " DB_COLUMNS " FROM \"%s\" WHERE bucket=? AND message_id>? ORDER BY message_id LIMIT 17"
|
|
: "SELECT " DB_COLUMNS " FROM \"%s\" WHERE bucket=? ORDER BY message_id LIMIT 17";
|
|
int rc = si_prep(si, &st, query), count = 0, bad = 0;
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)(prefix >> 39));
|
|
if (rc == SQLITE_OK && av) rc = sqlite3_bind_blob(st, 2, after->bytes, 32, SQLITE_STATIC);
|
|
if (rc != SQLITE_OK) { sqlite3_finalize(st); return db_error(si, "page prepare"); }
|
|
size_t used = 0; *more = 0; memset(next, 0, sizeof(*next));
|
|
while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
|
|
struct db_record r;
|
|
enum db_record_check check = db_read_record(si, st, &r);
|
|
if (check == DB_CHECK_VALID) check = db_verify_author_sig(si, &r);
|
|
if (check != DB_CHECK_VALID || !r.cache_valid) {
|
|
bad = check == DB_CHECK_BAD_SIGNATURE || check == DB_CHECK_BAD_FORMAT || (check == DB_CHECK_VALID && !r.cache_valid);
|
|
rc = SQLITE_ERROR; break;
|
|
}
|
|
size_t bytes = DB_WIRE_FIXED + r.data_len;
|
|
if (count == DB_BATCH_RECORDS || used + bytes > *len) {
|
|
if (!count) { rc = SQLITE_TOOBIG; break; }
|
|
*more = 1; break;
|
|
}
|
|
used += (size_t)db_encode_record(&r, buf + used); memcpy(next->bytes, r.message_id, 32); count++;
|
|
}
|
|
int end_rc = sqlite3_finalize(st);
|
|
if (bad) db_index_request(si);
|
|
if ((rc != SQLITE_DONE && !(rc == SQLITE_ROW && *more)) || end_rc != SQLITE_OK) return db_error(si, "page read");
|
|
*len = used; return 0;
|
|
}
|
|
|
|
static void db_done_sweep(struct DB_SYNC* db) {
|
|
if (db->dispatch_depth) return;
|
|
struct db_sync_done_cbk_entry** link = &db->done_cbks;
|
|
while (*link) {
|
|
struct db_sync_done_cbk_entry* e = *link;
|
|
if (e->removed) { *link = e->next; u_free(e); } else link = &e->next;
|
|
}
|
|
}
|
|
int db_sync_add_done_cbk(struct UTUN_INSTANCE* inst, db_sync_done_fn fn, void* arg) {
|
|
struct DB_SYNC* db = inst ? inst->db_sync : NULL;
|
|
if (!db || !fn) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history done subscription requires context/callback"); return -1; }
|
|
for (struct db_sync_done_cbk_entry* current = db->done_cbks; current; current = current->next)
|
|
if (!current->removed && current->fn == fn && current->arg == arg) return 0;
|
|
struct db_sync_done_cbk_entry* e = u_calloc(1, sizeof(*e));
|
|
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history done subscription allocation"); return -1; }
|
|
e->fn = fn; e->arg = arg; e->next = db->done_cbks; db->done_cbks = e;
|
|
return 0;
|
|
}
|
|
void db_sync_remove_done_cbk(struct UTUN_INSTANCE* inst, db_sync_done_fn fn, void* arg) {
|
|
struct DB_SYNC* db = inst ? inst->db_sync : NULL;
|
|
if (!db) return;
|
|
for (struct db_sync_done_cbk_entry* e = db->done_cbks; e; e = e->next)
|
|
if (!e->removed && e->fn == fn && e->arg == arg) { e->removed = 1; break; }
|
|
db_done_sweep(db);
|
|
}
|
|
static void db_message_result(void* ctx, const char* ns, uint64_t peer, int result) {
|
|
struct DB_SYNC_INSTANCE* si = db_sync_instance_find(ctx, strtoull(ns, NULL, 10));
|
|
if (!si) return;
|
|
struct SI_PEER* p = si_peer_add(si, peer);
|
|
if (!p) return;
|
|
p->sync_state = result == MT_OK ? 2 : 0;
|
|
if (result != MT_OK) { db_sync_resume_peer_check(si->db_sync); return; }
|
|
struct DB_SYNC* db = si->db_sync;
|
|
si->holds++; db->dispatch_depth++;
|
|
for (struct db_sync_done_cbk_entry* e = db->done_cbks; e && si->enabled; e = e->next)
|
|
if (!e->removed) e->fn(si, peer, e->arg);
|
|
db->dispatch_depth--; db_done_sweep(db); db_release(si);
|
|
}
|
|
|
|
static int db_message_peer(void* ctx, const char* ns, uint64_t peer) {
|
|
struct UTUN_INSTANCE* inst = ctx;
|
|
struct DB_SYNC_INSTANCE* si = db_sync_instance_find(inst, strtoull(ns, NULL, 10));
|
|
struct TOPO_GROUP* group = si ? topo_groups_find(inst->topo_groups, si->group_id) : NULL;
|
|
struct ETCP_CONN* conn = instance_find_conn(inst, peer);
|
|
return si && !si->sync_gated && !si->index.pending && topo_group_peer_ready(group, peer) && conn &&
|
|
member_sync_validate_pubkey(inst, ns, conn->crypto_ctx.peer_public_key);
|
|
}
|
|
static int db_sync_peer_throttled(struct DB_SYNC* db, struct SI_PEER* p) {
|
|
struct ETCP_CONN* conn = p && !p->sound_on ? instance_find_conn(db->inst, p->node_id) : NULL;
|
|
return conn && conn->peer_sleep_phase;
|
|
}
|
|
static int db_message_can_run(void* ctx, const char* ns, uint64_t peer) {
|
|
struct DB_SYNC_INSTANCE* si = db_sync_instance_find(ctx, strtoull(ns, NULL, 10));
|
|
if (!si || !db_message_peer(ctx, ns, peer)) return 0;
|
|
struct SI_PEER* p = si_peer_find(si, peer);
|
|
return !p || (!p->paused && !db_sync_peer_throttled(si->db_sync, p));
|
|
}
|
|
static int db_start_peer(struct DB_SYNC_INSTANCE* si, uint64_t peer) {
|
|
if (!merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES) || !db_message_can_run(si->db_sync->inst, si->ns, peer)) return 0;
|
|
struct SI_PEER* p = si_peer_add(si, peer);
|
|
if (!p) return -1;
|
|
if (merkle_sync_start(si->db_sync->inst, MT_DATA_MESSAGES, peer, si->ns, NULL, NULL) < 0) {
|
|
p->sync_state = 0; db_sync_resume_peer_check(si->db_sync); return -1;
|
|
}
|
|
p->sync_state = 1; db_sync_resume_peer_check(si->db_sync); return 0;
|
|
}
|
|
|
|
static void db_notice_unlink(struct db_notice* n) {
|
|
struct db_notice** link = &n->db->notices;
|
|
while (*link && *link != n) link = &(*link)->next;
|
|
if (*link) *link = n->next;
|
|
}
|
|
static void db_notice_run(void* arg) {
|
|
struct db_notice* n = arg; n->work = NULL; db_notice_unlink(n);
|
|
struct DB_SYNC_INSTANCE* si = db_instance_find(n->db, n->group_id);
|
|
if (si) {
|
|
si->holds++;
|
|
struct db_record* records = (struct db_record*)(n + 1);
|
|
for (unsigned i = 0; i < n->count && si->enabled; i++)
|
|
if (si->on_insert) si->on_insert(si, records[i].message_id, records[i].timestamp, records[i].data,
|
|
records[i].data_len, records[i].author, n->initial, records[i].updated, si->on_insert_arg);
|
|
db_release(si);
|
|
}
|
|
u_free(n);
|
|
}
|
|
static void db_notice_abort(struct db_notice* n) {
|
|
if (!n) return;
|
|
uasync_call_soon_cancel(n->db->inst->ua, n->work); db_notice_unlink(n); u_free(n);
|
|
}
|
|
static int db_notice_stage(struct DB_SYNC_INSTANCE* si, struct db_record* records, int count, const uint8_t* added, struct db_notice** out) {
|
|
size_t bodies = 0; unsigned total = 0; *out = NULL;
|
|
for (int i = 0; i < count; i++) if (added[i]) { total++; bodies += records[i].data_len; }
|
|
if (!total || !si->on_insert) return 0;
|
|
struct db_notice* n = u_calloc(1, sizeof(*n) + total * sizeof(struct db_record) + bodies);
|
|
if (!n) return db_error(si, "notification allocation");
|
|
n->db = si->db_sync; n->group_id = si->group_id; n->initial = 1; n->count = total;
|
|
struct db_record* saved = (struct db_record*)(n + 1); char* data = (char*)(saved + total); unsigned index = 0;
|
|
for (int i = 0; i < count; i++) if (added[i]) {
|
|
saved[index] = records[i]; saved[index].data = data; memcpy(data, records[i].data, records[i].data_len);
|
|
data += records[i].data_len; index++;
|
|
}
|
|
n->work = uasync_call_soon(si->db_sync->inst->ua, n, db_notice_run);
|
|
if (!n->work) { u_free(n); return db_error(si, "notification scheduling"); }
|
|
n->next = si->db_sync->notices; si->db_sync->notices = n; *out = n; return 0;
|
|
}
|
|
|
|
/* Полная проверка входящей страницы — до BEGIN; write-lock охватывает только небольшой commit данных и индекса. */
|
|
static int db_message_apply(void* ctx, const char* ns, uint64_t peer, uint64_t prefix, int av,
|
|
const merkle_key_t* after, const merkle_key_t* next_key, int more, const uint8_t* data, size_t len) {
|
|
(void)peer;
|
|
struct DB_SYNC_INSTANCE* si = db_sync_instance_find(ctx, strtoull(ns, NULL, 10));
|
|
if (!si || si->index.pending) return -1;
|
|
struct db_record records[DB_BATCH_RECORDS]; uint8_t added[DB_BATCH_RECORDS] = {0};
|
|
const uint8_t* p = data; int count = 0;
|
|
while (p < data + len) {
|
|
if (count == DB_BATCH_RECORDS || db_parse_record(si, &p, data + len, &records[count]) < 0) return -1;
|
|
struct db_record* r = &records[count];
|
|
if (merkle_sync_level_prefix(db_read64(r->message_id), MT_MAX_LEVEL) != prefix ||
|
|
(av && memcmp(r->message_id, after->bytes, 32) <= 0)) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history page key outside requested range ns=%s", ns); return -1;
|
|
}
|
|
if (db_verify_author_sig(si, r) != DB_CHECK_VALID) return -1;
|
|
if (count && memcmp(records[count - 1].message_id, r->message_id, 32) >= 0) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history incoming page not strictly ordered ns=%s", ns); return -1;
|
|
}
|
|
count++;
|
|
}
|
|
static const uint8_t empty[32];
|
|
if ((!count && (more || memcmp(next_key->bytes, empty, 32))) ||
|
|
(count && memcmp(next_key->bytes, records[count - 1].message_id, 32))) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history page cursor does not match records ns=%s", ns); return -1;
|
|
}
|
|
if (!count) return 0;
|
|
sqlite3* db = SI_DB(si);
|
|
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) return db_error(si, "page begin");
|
|
uint64_t next = si->next_id; int changed = 0; struct db_notice* notice = NULL;
|
|
for (int i = 0; i < count; i++) {
|
|
records[i].id = next;
|
|
int result = db_put_record(si, &records[i], DB_CHECK_VALID, NULL, &changed);
|
|
if (result < 0) goto rollback;
|
|
if (!result) { added[i] = 1; if (!records[i].updated) next++; }
|
|
}
|
|
if (db_notice_stage(si, records, count, added, ¬ice) < 0) goto rollback;
|
|
if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto rollback;
|
|
si->next_id = next;
|
|
if (changed) db_committed(si, changed);
|
|
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history page committed ns=%s records=%d changed=%d", ns, count, changed);
|
|
return 0;
|
|
rollback:
|
|
db_notice_abort(notice); db_rollback(si, "page apply"); return -1;
|
|
}
|
|
|
|
static const struct merkle_sync_data_ops db_message_ops = {
|
|
.update_bucket_hash = db_message_hash, .get_page = db_message_page, .apply_page = db_message_apply,
|
|
.validate_peer = db_message_peer, .can_run = db_message_can_run, .on_result = db_message_result
|
|
};
|
|
|
|
int db_sync_register_model(struct UTUN_INSTANCE* inst) {
|
|
if (!inst || !inst->db_sync || !inst->db_sync->enabled) return -1;
|
|
if (merkle_sync_registered(inst, MT_DATA_MESSAGES)) return 0;
|
|
if (merkle_sync_register(inst, MT_DATA_MESSAGES, &db_message_ops, inst) < 0) return -1;
|
|
for (int i = 0; i < inst->db_sync->instance_count; i++) db_index_request(inst->db_sync->instances[i]);
|
|
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history registered shared Merkle model");
|
|
return 0;
|
|
}
|
|
|
|
static int db_send(struct DB_SYNC_INSTANCE* si, uint64_t peer, uint8_t type, const uint8_t* data, size_t len) {
|
|
if (!db_message_can_run(si->db_sync->inst, si->ns, peer)) return -1;
|
|
struct ETCP_CONN* conn = instance_find_conn(si->db_sync->inst, peer);
|
|
if (!conn || !conn->links_up) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history transport unavailable peer=%016llx", peer); return -1; }
|
|
struct ll_entry* e = ll_alloc_lldgram((uint16_t)(10 + len));
|
|
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history packet allocation"); return -1; }
|
|
e->dgram[0] = ETCP_ID_DB_SYNC; db_write64(e->dgram + 1, si->group_id); e->dgram[9] = type;
|
|
if (len) memcpy(e->dgram + 10, data, len);
|
|
e->len = (uint16_t)(10 + len);
|
|
int rc = etcp_send(conn, e);
|
|
if (rc != 0) {
|
|
queue_dgram_free(e); queue_entry_free(e);
|
|
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history send rejected peer=%016llx rc=%d", peer, rc);
|
|
}
|
|
return rc;
|
|
}
|
|
|
|
static void db_emit_push(struct DB_SYNC_INSTANCE* si, const struct db_record* r, uint64_t source) {
|
|
si->holds++;
|
|
if (si->on_insert) si->on_insert(si, r->message_id, r->timestamp, r->data, r->data_len, r->author, 0, r->updated, si->on_insert_arg);
|
|
if (si->enabled) {
|
|
uint8_t wire[DB_WIRE_MAX]; size_t bytes = (size_t)db_encode_record(r, wire);
|
|
for (int i = 0; i < si->peer_count; i++)
|
|
if (si->peers[i].node_id != source && si->peers[i].sync_state && !si->peers[i].paused &&
|
|
!db_sync_peer_throttled(si->db_sync, &si->peers[i]))
|
|
db_send(si, si->peers[i].node_id, DB_MSG_PUSH, wire, bytes);
|
|
}
|
|
db_release(si);
|
|
}
|
|
static int db_insert_one(struct DB_SYNC_INSTANCE* si, struct db_record* r, const char* attrs) {
|
|
if (chat_message_id(si->group_id, r->timestamp, r->author, r->revision.nonce, r->message_id) < 0) return -1;
|
|
enum db_record_check incoming = db_verify_author_sig(si, r);
|
|
if (incoming == DB_CHECK_ERROR || incoming == DB_CHECK_DEFERRED) return -1;
|
|
sqlite3* db = SI_DB(si);
|
|
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) return db_error(si, "insert begin");
|
|
int changed = 0; r->id = si->next_id;
|
|
int result = db_put_record(si, r, incoming, attrs, &changed);
|
|
if (result == -1 || sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) { db_rollback(si, "insert"); return -1; }
|
|
if (!result && !r->updated) si->next_id++;
|
|
if (changed) db_committed(si, changed);
|
|
return result;
|
|
}
|
|
/* ACK конкретного сообщения: повтор не увеличивает счётчик, очередь отправки не считается доставкой. */
|
|
static int si_delivery_update(struct DB_SYNC_INSTANCE* si, const uint8_t message_id[32], uint64_t ver, uint64_t peer_id)
|
|
{
|
|
char hex[17]; snprintf(hex, sizeof(hex), "%016llx", (unsigned long long)peer_id);
|
|
sqlite3_stmt* stmt = NULL;
|
|
int rc = si_prep(si, &stmt,
|
|
"UPDATE \"%s\" SET flags=flags | CASE WHEN node_id=?3 THEN 1 ELSE 0 END,"
|
|
" delivered_peers=delivered_peers + (instr(',' || delivery_chain || ',', ',' || ?1 || ',')=0),"
|
|
" delivery_chain=CASE WHEN instr(',' || delivery_chain || ',', ',' || ?1 || ',')>0 THEN delivery_chain"
|
|
" WHEN delivery_chain='' THEN ?1 ELSE delivery_chain || ',' || ?1 END WHERE message_id=?2 AND ver=?4");
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_text(stmt, 1, hex, -1, SQLITE_STATIC);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(stmt, 2, message_id, 32, SQLITE_STATIC);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(stmt, 3, (sqlite3_int64)si->db_sync->inst->node_id);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(stmt, 4, (sqlite3_int64)ver);
|
|
if (rc == SQLITE_OK) rc = sqlite3_step(stmt);
|
|
int changed = sqlite3_changes(SI_DB(si)), final_rc = sqlite3_finalize(stmt);
|
|
if (rc != SQLITE_DONE || final_rc != SQLITE_OK) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history ACK update tbl=%s peer=%s rc=%d finalize=%d: %s",
|
|
SI_TBL(si), hex, rc, final_rc, sqlite3_errmsg(SI_DB(si))); return -1;
|
|
}
|
|
if (!changed) DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history ACK stale/unknown tbl=%s peer=%s ver=%llu",
|
|
SI_TBL(si), hex, (unsigned long long)ver);
|
|
else DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history ACK recorded tbl=%s peer=%s", SI_TBL(si), hex);
|
|
return 0;
|
|
}
|
|
|
|
|
|
static void db_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* e) {
|
|
if (!e) return;
|
|
struct UTUN_INSTANCE* inst = conn ? conn->instance : NULL;
|
|
struct DB_SYNC_INSTANCE* si = e->dgram && e->len >= 10 ? db_sync_instance_find(inst, db_read64(e->dgram + 1)) : NULL;
|
|
if (!si || !db_message_peer(inst, si->ns, conn->peer_node_id)) goto done;
|
|
const uint8_t* data = e->dgram + 10; size_t len = e->len - 10;
|
|
si->holds++;
|
|
if (e->dgram[9] == DB_MSG_ACK_PUSH) {
|
|
if (len == 40) si_delivery_update(si, data, db_read64(data + 32), conn->peer_node_id);
|
|
else DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history invalid ACK bytes=%zu", len);
|
|
} else if (e->dgram[9] == DB_MSG_PUSH) {
|
|
const uint8_t* p = data; struct db_record r;
|
|
if (db_parse_record(si, &p, data + len, &r) == 0 && p == data + len) {
|
|
int result = db_insert_one(si, &r, NULL);
|
|
if (!result) db_emit_push(si, &r, conn->peer_node_id);
|
|
if (result >= 0 && si->enabled) {
|
|
uint8_t ack[40]; memcpy(ack, r.message_id, 32); db_write64(ack + 32, r.revision.ver);
|
|
db_send(si, conn->peer_node_id, DB_MSG_ACK_PUSH, ack, sizeof(ack));
|
|
}
|
|
}
|
|
} else DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history unknown message type=%u", e->dgram[9]);
|
|
db_release(si);
|
|
done:
|
|
queue_dgram_free(e); queue_entry_free(e);
|
|
}
|
|
|
|
static uint64_t db_now_ms(struct DB_SYNC_INSTANCE* si) {
|
|
int64_t us;
|
|
if (si->db_sync->inst->ntp.synced) us = ntp_time_get_us(si->db_sync->inst);
|
|
else { struct timeval tv; utun_gettimeofday(&tv, NULL); us = (int64_t)tv.tv_sec * 1000000LL + tv.tv_usec; }
|
|
if (us < 0) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history wall clock before epoch"); return 0; }
|
|
return (uint64_t)us / 1000;
|
|
}
|
|
static void db_ttl_tick(void* arg) {
|
|
struct DB_SYNC_INSTANCE* si = arg; si->ttl_timer = NULL;
|
|
if (!si->enabled) return;
|
|
sqlite3* db = SI_DB(si); sqlite3_stmt* st = NULL;
|
|
uint64_t now = db_now_ms(si);
|
|
uint64_t ttl = (uint64_t)si->db_sync->inst->config->global.db_sync_ttl * 1000;
|
|
uint64_t cut = now > ttl ? now - ttl : 0;
|
|
int removed = 0, committed = 0;
|
|
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) { db_error(si, "TTL begin"); goto schedule; }
|
|
char query[512];
|
|
snprintf(query, sizeof(query), "DELETE FROM \"%s\" WHERE rowid IN (SELECT rowid FROM \"%s\""
|
|
" WHERE node_id=? AND (flags & 1)=0 AND CASE WHEN edited_at>0 THEN edited_at ELSE timestamp END<? LIMIT 16)", SI_TBL(si), SI_TBL(si));
|
|
int rc = sqlite3_prepare_v2(db, query, -1, &st, NULL);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)si->db_sync->inst->node_id);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)cut);
|
|
if (rc == SQLITE_OK) rc = sqlite3_step(st);
|
|
removed = sqlite3_changes(db);
|
|
int end_rc = sqlite3_finalize(st); st = NULL;
|
|
if (rc != SQLITE_DONE || end_rc != SQLITE_OK) { db_rollback(si, "TTL delete"); goto schedule; }
|
|
if (removed && merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES) && db_index_clear(si) < 0) {
|
|
db_rollback(si, "TTL index"); goto schedule;
|
|
}
|
|
if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) { db_rollback(si, "TTL commit"); goto schedule; }
|
|
committed = 1;
|
|
if (removed) {
|
|
db_index_request(si);
|
|
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history TTL committed tbl=%s removed=%d cutoff_ms=%llu",
|
|
SI_TBL(si), removed, (unsigned long long)cut);
|
|
}
|
|
schedule:
|
|
uint64_t delay = committed && removed == DB_BATCH_RECORDS ? 1 : DB_SYNC_TTL_INTERVAL * 10000ULL;
|
|
si->ttl_timer = uasync_set_timeout(si->db_sync->inst->ua, delay, si, db_ttl_tick, "history_ttl");
|
|
if (!si->ttl_timer) { db_error(si, "TTL timer"); db_sync_resume_peer_check(si->db_sync); }
|
|
}
|
|
|
|
static void db_sync_resume_peer_check(struct DB_SYNC* db) {
|
|
if (!db || !db->enabled || db->peer_check_timer) return;
|
|
#ifdef UTUN_HAVE_STANDBY
|
|
if (db->peer_check_wait) return;
|
|
#endif
|
|
db->peer_check_timer = uasync_set_timeout(db->inst->ua, DB_SYNC_PEER_CHECK_INTERVAL * 10000ULL, db, db_sync_peer_check_cb, "history_retry");
|
|
if (!db->peer_check_timer) DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history retry timer allocation");
|
|
}
|
|
#ifdef UTUN_HAVE_STANDBY
|
|
/* standby_wait может вызвать callback синхронно при отказе памяти: освобождаем handle и ставим таймер без рекурсии. */
|
|
static void db_sync_peer_check_wake(void* arg) {
|
|
struct DB_SYNC* db = arg; db->peer_check_wait = NULL;
|
|
db_sync_resume_peer_check(db);
|
|
}
|
|
#endif
|
|
static void db_sync_peer_check_cb(void* arg) {
|
|
struct DB_SYNC* db = arg; db->peer_check_timer = NULL; int pending = 0;
|
|
if (!db->enabled) return;
|
|
#ifdef UTUN_HAVE_STANDBY
|
|
db->peer_check_wait = NULL;
|
|
if (standby_get_sleep_tb() > 0) {
|
|
db->peer_check_wait = standby_wait(db, db_sync_peer_check_wake);
|
|
if (!db->peer_check_wait) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history standby wait allocation; retry later");
|
|
db_sync_resume_peer_check(db);
|
|
}
|
|
return;
|
|
}
|
|
#endif
|
|
for (int i = 0; i < db->instance_count; i++) {
|
|
struct DB_SYNC_INSTANCE* si = db->instances[i];
|
|
if (si->index.pending) {
|
|
pending = 1;
|
|
if (!si->index.timer) db_index_request(si);
|
|
}
|
|
for (int j = 0; j < si->peer_count; j++) {
|
|
struct SI_PEER* p = &si->peers[j];
|
|
if (p->sync_state != 2 && db_message_can_run(db->inst, si->ns, p->node_id)) {
|
|
pending = 1;
|
|
if (!p->sync_state) db_start_peer(si, p->node_id);
|
|
}
|
|
}
|
|
}
|
|
if (pending) db_sync_resume_peer_check(db);
|
|
}
|
|
static void db_sync_on_peer_ready(struct TOPO_GROUP* group, uint64_t peer, int ready, void* arg) {
|
|
(void)group; struct DB_SYNC_INSTANCE* si = arg;
|
|
if (!si->enabled) return;
|
|
struct SI_PEER* p = ready ? si_peer_add(si, peer) : si_peer_find(si, peer);
|
|
if (!p) return;
|
|
if (!ready) {
|
|
p->sync_state = 0;
|
|
merkle_sync_cancel(si->db_sync->inst, MT_DATA_MESSAGES, peer, si->ns);
|
|
} else db_start_peer(si, peer);
|
|
}
|
|
|
|
int db_sync_init(struct UTUN_INSTANCE* inst) {
|
|
if (!inst || inst->db_sync) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history invalid init"); return -1; }
|
|
struct DB_SYNC* db = u_calloc(1, sizeof(*db));
|
|
if (!db) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history allocation"); return -1; }
|
|
db->inst = inst; db->db = inst->topo_sqlite_db; inst->db_sync = db;
|
|
return inst->config->global.db_sync_enabled ? db_sync_enable(inst) : 0;
|
|
}
|
|
int db_sync_enable(struct UTUN_INSTANCE* inst) {
|
|
struct DB_SYNC* db = inst ? inst->db_sync : NULL;
|
|
if (!db || !inst->topo_sqlite_db) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history requires core SQLite"); return -1; }
|
|
if (db->enabled) return 0;
|
|
db->db = inst->topo_sqlite_db; db->enabled = 1;
|
|
etcp_bind(inst, ETCP_ID_DB_SYNC, db_sync_recv_cb);
|
|
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history enabled on shared core SQLite");
|
|
return 0;
|
|
}
|
|
struct DB_SYNC_INSTANCE* db_sync_instance_add(struct UTUN_INSTANCE* inst, const char* name, uint64_t group_id) {
|
|
struct DB_SYNC* db = inst ? inst->db_sync : NULL;
|
|
if (!db || !db->enabled || !si_name_valid(name) || !topo_groups_find(inst->topo_groups, group_id)) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history invalid table registration"); return NULL;
|
|
}
|
|
struct DB_SYNC_INSTANCE* si = db_instance_find(db, group_id);
|
|
if (si) return si;
|
|
if (db->instance_count == db->instance_capacity) {
|
|
int cap = db->instance_capacity ? db->instance_capacity * 2 : 4;
|
|
void* entries = u_realloc(db->instances, (size_t)cap * sizeof(*db->instances));
|
|
if (!entries) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history registry allocation"); return NULL; }
|
|
db->instances = entries; db->instance_capacity = cap;
|
|
}
|
|
si = u_calloc(1, sizeof(*si));
|
|
if (!si) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history table allocation"); return NULL; }
|
|
si->db_sync = db; si->group_id = group_id; si->enabled = 1; si->sync_gated = 1;
|
|
snprintf(si->table_name, sizeof(si->table_name), "%s", name);
|
|
snprintf(si->ns, sizeof(si->ns), "%llu", (unsigned long long)group_id);
|
|
db->instances[db->instance_count++] = si;
|
|
if (si_table_init(si) < 0) goto error;
|
|
char sql[512];
|
|
snprintf(sql, sizeof(sql), "CREATE INDEX IF NOT EXISTS \"idx_%s_ttl\" ON \"%s\"(node_id,timestamp);"
|
|
"CREATE INDEX IF NOT EXISTS \"idx_%s_bucket\" ON \"%s\"(bucket,message_id)", name, name, name, name);
|
|
if (sqlite3_exec(db->db, sql, NULL, NULL, NULL) != SQLITE_OK) { db_error(si, "create indexes"); goto error; }
|
|
sqlite3_stmt* st = NULL;
|
|
int rc = si_prep(si, &st, "SELECT COALESCE(MAX(id),0),COALESCE(MAX(timestamp),0) FROM \"%s\"");
|
|
if (rc == SQLITE_OK) rc = sqlite3_step(st);
|
|
if (rc == SQLITE_ROW) {
|
|
sqlite3_int64 max_id = sqlite3_column_int64(st, 0);
|
|
if (max_id < 0 || max_id == INT64_MAX) rc = SQLITE_TOOBIG;
|
|
else si->next_id = (uint64_t)max_id + 1;
|
|
si->last_timestamp_ms = (uint64_t)sqlite3_column_int64(st, 1);
|
|
}
|
|
int end_rc = sqlite3_finalize(st);
|
|
if (rc != SQLITE_ROW || end_rc != SQLITE_OK) { db_error(si, "load local sequence"); goto error; }
|
|
if (topo_group_add_peer_ready_cbk(topo_groups_find(inst->topo_groups, group_id), db_sync_on_peer_ready, si) < 0) goto error;
|
|
db_index_request(si);
|
|
si->ttl_timer = uasync_set_timeout(inst->ua, DB_SYNC_TTL_INTERVAL * 10000ULL, si, db_ttl_tick, "history_ttl");
|
|
if (!si->ttl_timer) { db_error(si, "initial TTL timer"); goto error; }
|
|
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history table registered tbl=%s group=%s; preflight pending", name, si->ns);
|
|
return si;
|
|
error:
|
|
db_sync_instance_remove(si); return NULL;
|
|
}
|
|
void db_sync_instance_remove(struct DB_SYNC_INSTANCE* si) {
|
|
if (!si || !si->enabled) return;
|
|
si->enabled = 0; struct DB_SYNC* db = si->db_sync;
|
|
topo_group_remove_peer_ready_cbk(topo_groups_find(db->inst->topo_groups, si->group_id), db_sync_on_peer_ready, si);
|
|
merkle_sync_cancel_dataset_ns(db->inst, MT_DATA_MESSAGES, si->ns);
|
|
if (si->index.timer) uasync_cancel_timeout(db->inst->ua, si->index.timer);
|
|
if (si->ttl_timer) uasync_cancel_timeout(db->inst->ua, si->ttl_timer);
|
|
for (struct db_notice* n = db->notices; n;) {
|
|
struct db_notice* next = n->next; if (n->group_id == si->group_id) db_notice_abort(n); n = next;
|
|
}
|
|
for (int i = 0; i < db->instance_count; i++) if (db->instances[i] == si) {
|
|
memmove(db->instances + i, db->instances + i + 1, (size_t)(db->instance_count - i - 1) * sizeof(*db->instances));
|
|
db->instance_count--; break;
|
|
}
|
|
u_free(si->peers); si->peers = NULL; si->peer_count = 0;
|
|
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history table detached tbl=%s; signed records preserved", SI_TBL(si));
|
|
if (!si->holds) u_free(si);
|
|
}
|
|
void db_sync_destroy(struct UTUN_INSTANCE* inst) {
|
|
struct DB_SYNC* db = inst ? inst->db_sync : NULL;
|
|
if (!db) return;
|
|
db->enabled = 0;
|
|
if (db->peer_check_timer) uasync_cancel_timeout(inst->ua, db->peer_check_timer);
|
|
#ifdef UTUN_HAVE_STANDBY
|
|
if (db->peer_check_wait) standby_wait_cancel(db->peer_check_wait);
|
|
#endif
|
|
while (db->instance_count) db_sync_instance_remove(db->instances[db->instance_count - 1]);
|
|
while (db->notices) db_notice_abort(db->notices);
|
|
while (db->done_cbks) {
|
|
struct db_sync_done_cbk_entry* e = db->done_cbks; db->done_cbks = e->next; u_free(e);
|
|
}
|
|
etcp_unbind(inst, ETCP_ID_DB_SYNC);
|
|
inst->db_sync = NULL; u_free(db->instances); u_free(db);
|
|
}
|
|
|
|
int db_sync_insert_signed(struct DB_SYNC_INSTANCE* si, const char* data, size_t len, const uint8_t* sig, size_t sig_len,
|
|
uint64_t timestamp, const char* attrs, const struct db_message_revision* revision) {
|
|
if (!si || !si->enabled || !sig || sig_len != 64 || !revision) {
|
|
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history invalid insertion"); return -1;
|
|
}
|
|
struct db_record r = {.timestamp = timestamp, .author = si->db_sync->inst->node_id, .data = data, .data_len = len, .sig_len = sig_len};
|
|
r.revision = *revision;
|
|
memcpy(r.signature, sig, 64);
|
|
int result = db_insert_one(si, &r, attrs);
|
|
if (!result) db_emit_push(si, &r, r.author);
|
|
return result;
|
|
}
|
|
uint32_t db_sync_count(struct DB_SYNC_INSTANCE* si) {
|
|
if (!si || !si->enabled) return 0;
|
|
sqlite3_stmt* st = NULL; int rc = si_prep(si, &st, "SELECT COUNT(*) FROM \"%s\"");
|
|
if (rc == SQLITE_OK) rc = sqlite3_step(st);
|
|
uint32_t count = rc == SQLITE_ROW ? (uint32_t)sqlite3_column_int64(st, 0) : 0;
|
|
if (rc != SQLITE_ROW) db_error(si, "count"); sqlite3_finalize(st); return count;
|
|
}
|
|
uint64_t db_sync_get_last_timestamp(struct DB_SYNC_INSTANCE* si) { return si ? si->last_timestamp_ms : 0; }
|
|
uint64_t db_sync_next_timestamp(struct DB_SYNC_INSTANCE* si) {
|
|
if (!si) return 0; uint64_t now = db_now_ms(si);
|
|
if (now <= si->last_timestamp_ms) now = si->last_timestamp_ms + 1;
|
|
return si->last_timestamp_ms = now;
|
|
}
|
|
void db_sync_set_insert_cb(struct DB_SYNC_INSTANCE* si, db_sync_insert_cb fn, void* arg) {
|
|
if (si) { si->on_insert = fn; si->on_insert_arg = arg; }
|
|
}
|
|
int db_sync_select(struct DB_SYNC_INSTANCE* si, uint32_t offset, uint32_t limit, db_sync_select_cb cb, void* arg) {
|
|
if (!si || !si->enabled || !cb) return -1;
|
|
sqlite3_stmt* st = NULL; si->holds++;
|
|
int rc = si_prep(si, &st, "SELECT id,timestamp,node_id,data,author_signature,delivered_peers,delivery_chain"
|
|
" FROM \"%s\" ORDER BY timestamp,message_id LIMIT ? OFFSET ?");
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, limit ? limit : -1LL);
|
|
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, offset);
|
|
int count = 0;
|
|
if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW && si->enabled) {
|
|
const char* data = sqlite3_column_blob(st, 3); int bytes = sqlite3_column_bytes(st, 3);
|
|
const uint8_t* signature = sqlite3_column_blob(st, 4); int sig_len = sqlite3_column_bytes(st, 4);
|
|
const char* chain = (const char*)sqlite3_column_text(st, 6);
|
|
if (sqlite3_errcode(SI_DB(si)) == SQLITE_NOMEM) { rc = SQLITE_NOMEM; break; }
|
|
cb(arg, (uint64_t)sqlite3_column_int64(st, 0), (uint64_t)sqlite3_column_int64(st, 1), data, (size_t)bytes,
|
|
(uint64_t)sqlite3_column_int64(st, 2), signature, (size_t)sig_len, sqlite3_column_int(st, 5), chain ? chain : "");
|
|
count++;
|
|
}
|
|
int end_rc = sqlite3_finalize(st);
|
|
if ((rc != SQLITE_DONE && si->enabled) || end_rc != SQLITE_OK) { db_error(si, "select"); count = -1; }
|
|
db_release(si); return count;
|
|
}
|
|
|
|
/* Диагностический синхронный обход: в рабочем обмене preflight идёт порциями по таймеру. */
|
|
int db_sync_verify(struct DB_SYNC_INSTANCE* si) {
|
|
if (!si || !si->enabled || !merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES)) return -1;
|
|
if (!si->index.pending) db_index_request(si);
|
|
while (si->index.pending) if (db_index_batch(si) != 0) return si->index.waiting_key ? 1 : -1;
|
|
if (si->index.timer) { uasync_cancel_timeout(si->db_sync->inst->ua, si->index.timer); si->index.timer = NULL; }
|
|
uint8_t root[32]; return merkle_sync_read_hash(si->db_sync->inst, MT_DATA_MESSAGES, si->ns, 0, 0, root);
|
|
}
|
|
int db_sync_read_root(struct DB_SYNC_INSTANCE* si, uint8_t root[32]) {
|
|
if (!si || !si->enabled || !root) return -1;
|
|
if (si->index.pending) return 1;
|
|
return merkle_sync_read_hash(si->db_sync->inst, MT_DATA_MESSAGES, si->ns, 0, 0, root);
|
|
}
|
|
void db_sync_peer_set_paused(struct DB_SYNC_INSTANCE* si, uint64_t peer, int paused) {
|
|
struct SI_PEER* p = si ? si_peer_add(si, peer) : NULL;
|
|
if (!p) return;
|
|
p->paused = !!paused; p->sync_state = 0;
|
|
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history peer pause tbl=%s peer=%016llx paused=%u",
|
|
SI_TBL(si), (unsigned long long)peer, p->paused);
|
|
if (paused) merkle_sync_cancel(si->db_sync->inst, MT_DATA_MESSAGES, peer, si->ns);
|
|
else db_start_peer(si, peer);
|
|
}
|
|
void db_sync_reinitiate(struct DB_SYNC_INSTANCE* si, uint64_t peer) {
|
|
if (!si) return;
|
|
struct SI_PEER* p = si_peer_add(si, peer); if (!p) return;
|
|
p->paused = 0; p->sync_state = 0; db_start_peer(si, peer);
|
|
}
|
|
void db_sync_instance_set_gated(struct DB_SYNC_INSTANCE* si, int gated) {
|
|
if (!si || !si->enabled) return;
|
|
si->sync_gated = !!gated;
|
|
if (gated) merkle_sync_cancel_dataset_ns(si->db_sync->inst, MT_DATA_MESSAGES, si->ns);
|
|
struct TOPO_GROUP* group = topo_groups_find(si->db_sync->inst->topo_groups, si->group_id);
|
|
for (struct ll_entry* e = group && group->senders_list ? group->senders_list->head : NULL; e; e = e->next) {
|
|
struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data;
|
|
struct SI_PEER* p = si_peer_add(si, item->node_id); if (p) p->sync_state = 0;
|
|
if (!gated) db_start_peer(si, item->node_id);
|
|
}
|
|
db_sync_resume_peer_check(si->db_sync);
|
|
}
|
|
void db_sync_peer_policy_changed(struct UTUN_INSTANCE* inst, uint64_t peer) {
|
|
struct DB_SYNC* db = inst ? inst->db_sync : NULL; if (!db) return;
|
|
merkle_sync_recheck_peer(inst, peer);
|
|
for (int i = 0; i < db->instance_count; i++) {
|
|
struct DB_SYNC_INSTANCE* si = db->instances[i]; struct SI_PEER* p = si_peer_find(si, peer);
|
|
if (p && !p->sync_state) db_start_peer(si, peer);
|
|
}
|
|
}
|
|
void db_sync_set_peer_sound(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, int sound) {
|
|
struct DB_SYNC_INSTANCE* si = db_sync_instance_find(inst, group);
|
|
struct SI_PEER* p = si ? si_peer_add(si, peer) : NULL;
|
|
if (p) { p->sound_on = !!sound; db_sync_peer_policy_changed(inst, peer); }
|
|
}
|
|
void db_sync_members_changed(struct UTUN_INSTANCE* inst, const char* ns) {
|
|
struct DB_SYNC* db = inst ? inst->db_sync : NULL; if (!db) return;
|
|
merkle_sync_revalidate_peers(inst, MT_DATA_MESSAGES, ns);
|
|
struct DB_SYNC_INSTANCE* channel = db_sync_instance_find(inst, strtoull(ns, NULL, 10));
|
|
if (channel) {
|
|
for (int i = 0; i < channel->peer_count; i++)
|
|
if (!channel->peers[i].sync_state) db_start_peer(channel, channel->peers[i].node_id);
|
|
}
|
|
for (int i = 0; i < db->instance_count; i++) {
|
|
struct DB_SYNC_INSTANCE* si = db->instances[i];
|
|
if (si->index.pending && si->index.waiting_key) {
|
|
if (si->index.timer) uasync_cancel_timeout(inst->ua, si->index.timer);
|
|
si->index.timer = uasync_set_timeout(inst->ua, 0, si, db_index_tick, "history_key_ready");
|
|
if (!si->index.timer) db_sync_resume_peer_check(db);
|
|
}
|
|
}
|
|
}
|