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

Add signed versioned feedback storage and Merkle adapter

master
evgeny 5 days ago
parent
864a250eea
commit
641a8318a8
1 changed files with 425 additions and 0 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 425
      src/chat/chat_feedback.c

425
src/chat/chat_feedback.c
Unescape View File

@ -0,0 +1,425 @@
#include "chat_feedback.h"
#include "merkle_sync.h"
#include "member_sync.h"
#include "chat_event.h"
#include "../utun_instance.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "../transport_layer/secure_channel.h"
#include "../../lib/platform_compat.h"
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <errno.h>
#define FB_FIXED_SIZE 115
#define FB_WIRE_MAX (FB_FIXED_SIZE + CHAT_FEEDBACK_DATA_MAX)
static void fb_write64(uint8_t* p, uint64_t n) { n = htobe64(n); memcpy(p, &n, 8); }
static uint64_t fb_read64(const uint8_t* p) { uint64_t n; memcpy(&n, p, 8); return be64toh(n); }
/* SQL-имя строится только из канонического числового ID. */
static int fb_table(struct UTUN_INSTANCE* inst, const char* channel, char table[40], uint64_t* id) {
if (!inst || !inst->topo_sqlite_db || !channel || !channel[0] || strlen(channel) > 19 || channel[0] == '0') goto invalid;
for (const char* p = channel; *p; p++) if (*p < '0' || *p > '9') goto invalid;
errno = 0; char* end;
uint64_t n = strtoull(channel, &end, 10);
if (errno || *end || n > INT64_MAX) goto invalid;
snprintf(table, 40, "feedback_%s", channel); if (id) *id = n;
return 0;
invalid:
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: invalid channel=%s", channel ? channel : "-"); return -1;
}
/* Общая диагностика ошибок SQLite. */
static int fb_db_error(struct UTUN_INSTANCE* inst, const char* channel, const char* op) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: %s ch=%s: %s", op, channel,
inst && inst->topo_sqlite_db ? sqlite3_errmsg(inst->topo_sqlite_db) : "no DB");
return -1;
}
/* Канонический ID сообщения не зависит от локального порядкового номера. */
int chat_message_id(uint64_t channel_id, uint64_t timestamp, const uint8_t signature[64], uint8_t id[32]) {
static const char domain[] = "utun-chat-message-v1";
uint8_t body[sizeof(domain) - 1 + 16 + 64];
if (!channel_id || !signature || !id) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: invalid message identity arguments"); return -1;
}
memcpy(body, domain, sizeof(domain) - 1);
fb_write64(body + sizeof(domain) - 1, channel_id);
fb_write64(body + sizeof(domain) - 1 + 8, timestamp);
memcpy(body + sizeof(domain) - 1 + 16, signature, 64);
if (EVP_Digest(body, sizeof(body), id, NULL, EVP_sha256(), NULL) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: message identity digest failed"); return -1;
}
return 0;
}
/* Каноническая запись: message:32,node:8,kind:1,ver:8,len:2,data:len,signature:64. */
static int fb_encode(const struct chat_feedback_record* r, uint8_t* wire) {
if (!r || !r->node_id || r->node_id > INT64_MAX || !r->ver || r->ver > INT64_MAX
|| (r->kind != CHAT_FEEDBACK_REACTION && r->kind != CHAT_FEEDBACK_ANSWER) || r->data_len > CHAT_FEEDBACK_DATA_MAX) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: invalid record fields"); return -1;
}
memcpy(wire, r->message_id, 32); fb_write64(wire + 32, r->node_id); wire[40] = r->kind;
fb_write64(wire + 41, r->ver); wire[49] = r->data_len >> 8; wire[50] = r->data_len & 255;
memcpy(wire + 51, r->data, r->data_len); memcpy(wire + 51 + r->data_len, r->signature, 64);
return FB_FIXED_SIZE + r->data_len;
}
/* Читает ровно одну запись без предположений о выравнивании входного буфера. */
static int fb_decode(const uint8_t* wire, size_t len, struct chat_feedback_record* r) {
if (len < FB_FIXED_SIZE) goto invalid;
uint16_t bytes = ((uint16_t)wire[49] << 8) | wire[50];
if (bytes > CHAT_FEEDBACK_DATA_MAX || len < FB_FIXED_SIZE + (size_t)bytes) goto invalid;
memset(r, 0, sizeof(*r)); memcpy(r->message_id, wire, 32); r->node_id = fb_read64(wire + 32);
r->kind = wire[40]; r->ver = fb_read64(wire + 41); r->data_len = bytes;
memcpy(r->data, wire + 51, bytes); memcpy(r->signature, wire + 51 + bytes, 64);
if (!r->node_id || r->node_id > INT64_MAX || !r->ver || r->ver > INT64_MAX
|| (r->kind != CHAT_FEEDBACK_REACTION && r->kind != CHAT_FEEDBACK_ANSWER)) goto invalid;
return FB_FIXED_SIZE + bytes;
invalid:
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: malformed record bytes=%zu", len); return -1;
}
/* Подпись связывает канал с записью; ключ дерева связывает неизменяемую тройку. */
static int fb_signed_bytes(uint64_t channel, const struct chat_feedback_record* r, uint8_t* out) {
static const char domain[] = "utun-chat-feedback-v1";
memcpy(out, domain, sizeof(domain) - 1); fb_write64(out + sizeof(domain) - 1, channel);
int n = fb_encode(r, out + sizeof(domain) - 1 + 8);
return n < 0 ? -1 : (int)(sizeof(domain) - 1 + 8) + n - 64;
}
static int fb_key(uint64_t channel, const struct chat_feedback_record* r, merkle_key_t* key) {
uint8_t identity[49]; fb_write64(identity, channel); memcpy(identity + 8, r->message_id, 32);
fb_write64(identity + 40, r->node_id); identity[48] = r->kind;
if (EVP_Digest(identity, sizeof(identity), key->bytes, NULL, EVP_sha256(), NULL) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: record key digest failed"); return -1;
}
return 0;
}
/* Членство и привязка Ed25519 к X25519 подтверждаются существующими подписями мембера. */
static int fb_author_key(struct UTUN_INSTANCE* inst, const char* channel, uint64_t author, uint8_t key[32]) {
char sql[384];
snprintf(sql, sizeof(sql), "SELECT x25519_pubkey,ed25519_pubkey,signed_by,signature,join_sig,join_ts"
" FROM peers_%s WHERE node_id=? AND source=0", channel);
sqlite3_stmt* st = NULL; sqlite3* db = inst->topo_sqlite_db;
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) return fb_db_error(inst, channel, "author lookup");
sqlite3_bind_int64(st, 1, (sqlite3_int64)author);
int rc = sqlite3_step(st), ok = 0;
if (rc == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 32 && sqlite3_column_bytes(st, 1) == 32
&& sqlite3_column_bytes(st, 3) == 64 && sqlite3_column_bytes(st, 4) == 64 && sqlite3_column_int64(st, 5) > 0) {
struct ms_member_rec member = { .node_id = author, .x25519 = sqlite3_column_blob(st, 0),
.ed25519 = sqlite3_column_blob(st, 1), .signed_by = (uint64_t)sqlite3_column_int64(st, 2),
.signature = sqlite3_column_blob(st, 3), .join_sig = sqlite3_column_blob(st, 4),
.join_ts = (uint64_t)sqlite3_column_int64(st, 5) };
uint8_t x[32], ed[32], body[112];
int n = topo_node_sqlite_channel_get(db, channel, NULL, 0, NULL, x, ed, NULL);
if (n == 0) n = member_sync_build_join_msg(x, ed, author, member.x25519, member.join_ts, body, sizeof(body));
ok = n == 112 && sc_derive_node_id_from_pubkey(member.x25519) == author
&& member_sync_verify_signature(db, channel, &member) == 1
&& sc_ed25519_verify(member.ed25519, body, n, member.join_sig) == SC_OK;
if (ok) memcpy(key, member.ed25519, 32);
}
sqlite3_finalize(st);
if (ok) return 0;
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "feedback: author not verified ch=%s node=%016llx lookup=%d",
channel, (unsigned long long)author, rc); return -1;
}
/* Создаёт одну таблицу подписанных состояний и индекс обхода листьев. */
int chat_feedback_ensure_channel(struct UTUN_INSTANCE* inst, const char* channel) {
char table[40], sql[1024]; uint64_t id;
if (fb_table(inst, channel, table, &id) < 0) return -1;
if (topo_node_sqlite_channel_get(inst->topo_sqlite_db, channel, NULL, 0, NULL, NULL, NULL, NULL) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: unknown channel=%s", channel); return -1;
}
snprintf(sql, sizeof(sql), "CREATE TABLE IF NOT EXISTS %s("
"record_key BLOB PRIMARY KEY CHECK(length(record_key)=32),"
"bucket INTEGER NOT NULL,message_id BLOB NOT NULL CHECK(length(message_id)=32),"
"node_id INTEGER NOT NULL,kind INTEGER NOT NULL,ver INTEGER NOT NULL CHECK(ver>0),"
"data BLOB NOT NULL,signature BLOB NOT NULL CHECK(length(signature)=64),"
"UNIQUE(message_id,node_id,kind));"
"CREATE INDEX IF NOT EXISTS idx_%s_bucket ON %s(bucket,record_key)", table, table, table);
if (sqlite3_exec(inst->topo_sqlite_db, sql, NULL, NULL, NULL) != SQLITE_OK) return fb_db_error(inst, channel, "ensure table");
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "feedback: table ready ch=%s table=%s", channel, table); return 0;
}
/* Преобразует проверенные столбцы SQLite в самостоятельную запись. */
static int fb_read_row(sqlite3_stmt* st, struct chat_feedback_record* r) {
int bytes = sqlite3_column_bytes(st, 4);
if (sqlite3_column_bytes(st, 0) != 32 || bytes < 0 || bytes > CHAT_FEEDBACK_DATA_MAX
|| sqlite3_column_bytes(st, 5) != 64) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: corrupt local record"); return -1;
}
memset(r, 0, sizeof(*r)); memcpy(r->message_id, sqlite3_column_blob(st, 0), 32);
r->node_id = (uint64_t)sqlite3_column_int64(st, 1); r->kind = sqlite3_column_int(st, 2);
r->ver = (uint64_t)sqlite3_column_int64(st, 3); r->data_len = bytes;
if (bytes) memcpy(r->data, sqlite3_column_blob(st, 4), bytes);
memcpy(r->signature, sqlite3_column_blob(st, 5), 64); return 0;
}
int chat_feedback_get(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message_id[32],
uint64_t node_id, uint8_t kind, struct chat_feedback_record* out) {
char table[40], sql[256];
if (fb_table(inst, channel, table, NULL) < 0 || !message_id || !out) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: invalid lookup arguments"); return -1;
}
snprintf(sql, sizeof(sql), "SELECT message_id,node_id,kind,ver,data,signature FROM %s"
" WHERE message_id=? AND node_id=? AND kind=?", table);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL) != SQLITE_OK) return fb_db_error(inst, channel, "get");
sqlite3_bind_blob(st, 1, message_id, 32, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)node_id);
sqlite3_bind_int(st, 3, kind);
int rc = sqlite3_step(st), result = rc == SQLITE_ROW ? (fb_read_row(st, out) < 0 ? -1 : 1) : 0;
sqlite3_finalize(st);
if (rc != SQLITE_ROW && rc != SQLITE_DONE) return fb_db_error(inst, channel, "get step");
return result;
}
/* Применяет внутри транзакции; уведомления строго после commit вызывающего. */
static int fb_apply_row(struct UTUN_INSTANCE* inst, const char* channel, const struct chat_feedback_record* r) {
char table[40], sql[512]; uint64_t id; uint8_t body[FB_WIRE_MAX + 32], key[32];
if (fb_table(inst, channel, table, &id) < 0) return -1;
int bytes = fb_signed_bytes(id, r, body);
if (bytes < 0 || fb_author_key(inst, channel, r->node_id, key) < 0) return -1;
if (sc_ed25519_verify(key, body, bytes, r->signature) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: invalid signature ch=%s node=%016llx kind=%u ver=%llu",
channel, (unsigned long long)r->node_id, r->kind, (unsigned long long)r->ver); return -1;
}
struct chat_feedback_record old;
int found = chat_feedback_get(inst, channel, r->message_id, r->node_id, r->kind, &old);
if (found < 0) return -1;
if (found) {
int cmp = memcmp(r->data, old.data, r->data_len < old.data_len ? r->data_len : old.data_len);
if (!cmp) cmp = (int)r->data_len - old.data_len;
if (r->ver == old.ver && cmp) DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC,
"feedback: same-version conflict ch=%s node=%016llx kind=%u ver=%llu incoming_wins=%d",
channel, (unsigned long long)r->node_id, r->kind, (unsigned long long)r->ver, cmp > 0);
if (r->ver < old.ver || (r->ver == old.ver && cmp <= 0)) {
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "feedback: unchanged ch=%s node=%016llx kind=%u ver=%llu current=%llu",
channel, (unsigned long long)r->node_id, r->kind, (unsigned long long)r->ver, (unsigned long long)old.ver);
return 0;
}
}
merkle_key_t record_key; if (fb_key(id, r, &record_key) < 0) return -1;
snprintf(sql, sizeof(sql), "INSERT INTO %s(record_key,bucket,message_id,node_id,kind,ver,data,signature)"
" VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(record_key) DO UPDATE SET ver=excluded.ver,data=excluded.data,signature=excluded.signature", table);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL) != SQLITE_OK) return fb_db_error(inst, channel, "put");
sqlite3_bind_blob(st, 1, record_key.bytes, 32, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)(fb_read64(record_key.bytes) >> 39));
sqlite3_bind_blob(st, 3, r->message_id, 32, SQLITE_STATIC); sqlite3_bind_int64(st, 4, (sqlite3_int64)r->node_id);
sqlite3_bind_int(st, 5, r->kind); sqlite3_bind_int64(st, 6, (sqlite3_int64)r->ver);
sqlite3_bind_blob(st, 7, r->data, r->data_len, SQLITE_STATIC); sqlite3_bind_blob(st, 8, r->signature, 64, SQLITE_STATIC);
int rc = sqlite3_step(st); sqlite3_finalize(st);
if (rc != SQLITE_DONE) return fb_db_error(inst, channel, "put step");
if (merkle_sync_recompute_path(inst, MT_DATA_FEEDBACK, channel, fb_read64(record_key.bytes)) < 0) return -1;
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "feedback: applied ch=%s node=%016llx kind=%u ver=%llu previous=%llu bytes=%u",
channel, (unsigned long long)r->node_id, r->kind, (unsigned long long)r->ver,
(unsigned long long)(found ? old.ver : 0), r->data_len); return 1;
}
/* Событие указывает конкретное сообщение; данные SQLite уже опубликованы. */
static void fb_committed(struct UTUN_INSTANCE* inst, const char* channel, const struct chat_feedback_record* r) {
uint8_t event[54]; size_t len = strlen(channel); event[0] = len;
memcpy(event + 1, channel, len); memcpy(event + 1 + len, r->message_id, 32); event[33 + len] = r->kind;
chat_event_post(inst, CHAT_EVT_FEEDBACK_CHANGED, event, (int)len + 34);
}
/* Завершает savepoint, при ошибке не публикует частичные изменения. */
static int fb_finish_transaction(struct UTUN_INSTANCE* inst, const char* channel, int result) {
sqlite3* db = inst->topo_sqlite_db;
if (result < 0 && sqlite3_exec(db, "ROLLBACK TO feedback_update", NULL, NULL, NULL) != SQLITE_OK)
fb_db_error(inst, channel, "rollback");
if (sqlite3_exec(db, "RELEASE feedback_update", NULL, NULL, NULL) == SQLITE_OK) return result;
fb_db_error(inst, channel, "commit");
if (sqlite3_exec(db, "ROLLBACK TO feedback_update; RELEASE feedback_update", NULL, NULL, NULL) != SQLITE_OK)
fb_db_error(inst, channel, "commit cleanup");
return -1;
}
int chat_feedback_apply(struct UTUN_INSTANCE* inst, const char* channel, const struct chat_feedback_record* r) {
char table[40];
if (fb_table(inst, channel, table, NULL) < 0 || !r) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: invalid apply arguments"); return -1;
}
if (!sqlite3_get_autocommit(inst->topo_sqlite_db)) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: public apply inside external transaction ch=%s", channel); return -1;
}
if (sqlite3_exec(inst->topo_sqlite_db, "SAVEPOINT feedback_update", NULL, NULL, NULL) != SQLITE_OK)
return fb_db_error(inst, channel, "begin");
int result = fb_finish_transaction(inst, channel, fb_apply_row(inst, channel, r));
if (result > 0) { merkle_sync_changed(inst, MT_DATA_FEEDBACK, channel); fb_committed(inst, channel, r); }
return result;
}
int chat_feedback_set(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message_id[32],
uint8_t kind, const uint8_t* data, size_t len, uint64_t* out_ver) {
char table[40]; uint64_t id;
if (fb_table(inst, channel, table, &id) < 0 || !message_id || len > CHAT_FEEDBACK_DATA_MAX || (len && !data)) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: invalid local state"); return -1;
}
struct chat_feedback_record r = { .node_id = inst->node_id, .ver = 1, .kind = kind, .data_len = len }, old;
memcpy(r.message_id, message_id, 32); if (len) memcpy(r.data, data, len);
int found = chat_feedback_get(inst, channel, message_id, inst->node_id, kind, &old);
if (found < 0) return -1;
if (found && old.data_len == len && !memcmp(old.data, r.data, len)) { if (out_ver) *out_ver = old.ver; return 0; }
if (found && old.ver == INT64_MAX) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: version exhausted ch=%s", channel); return -1;
}
if (found) r.ver = old.ver + 1;
uint8_t body[FB_WIRE_MAX + 32]; int bytes = fb_signed_bytes(id, &r, body);
if (bytes < 0 || sc_ed25519_sign(inst->my_ed25519_privkey, body, bytes, r.signature) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: local signing failed ch=%s", channel); return -1;
}
int result = chat_feedback_apply(inst, channel, &r);
if (result >= 0 && out_ver) *out_ver = r.ver;
if (result > 0) DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "feedback: local state saved ch=%s kind=%u ver=%llu bytes=%zu",
channel, kind, (unsigned long long)r.ver, len);
return result;
}
/* Читает лист в полном лексикографическом порядке record_key. */
static int fb_page(void* ctx, const char* channel, uint64_t prefix, int av, const merkle_key_t* after,
uint8_t* data, size_t* len, merkle_key_t* next, int* more) {
struct UTUN_INSTANCE* inst = ctx; char table[40], sql[384];
if (fb_table(inst, channel, table, NULL) < 0 || *len < 2) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: invalid page capacity"); return -1;
}
merkle_key_t cursor = *after; memset(next, 0, sizeof(*next)); *more = 0;
snprintf(sql, sizeof(sql), "SELECT message_id,node_id,kind,ver,data,signature,record_key FROM %s"
" WHERE bucket=? AND (?=0 OR record_key>?) ORDER BY record_key", table);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL) != SQLITE_OK) return fb_db_error(inst, channel, "page");
sqlite3_bind_int64(st, 1, (sqlite3_int64)(prefix >> 39)); sqlite3_bind_int(st, 2, av);
sqlite3_bind_blob(st, 3, cursor.bytes, 32, SQLITE_STATIC);
size_t off = 2; unsigned count = 0; int step, result = 0;
while ((step = sqlite3_step(st)) == SQLITE_ROW) {
struct chat_feedback_record r; uint8_t wire[FB_WIRE_MAX];
if (fb_read_row(st, &r) < 0 || sqlite3_column_bytes(st, 6) != 32) { result = -1; break; }
int bytes = fb_encode(&r, wire); if (bytes < 0) { result = -1; break; }
if (off + bytes > *len) {
if (!count) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: record exceeds page capacity ch=%s", channel); result = -1; }
else *more = 1;
break;
}
memcpy(data + off, wire, bytes); off += bytes; count++;
memcpy(next->bytes, sqlite3_column_blob(st, 6), 32);
}
if (result == 0 && !*more && step != SQLITE_DONE) result = fb_db_error(inst, channel, "page step");
sqlite3_finalize(st); data[0] = count >> 8; data[1] = count & 255; *len = off; return result;
}
/* Хеширует подписанные записи листа в том же порядке, что и страничный обход. */
static int fb_hash(void* ctx, const char* channel, uint8_t level, uint64_t prefix, EVP_MD_CTX* hash) {
struct UTUN_INSTANCE* inst = ctx; char table[40], sql[256];
if (fb_table(inst, channel, table, NULL) < 0 || level != MT_MAX_LEVEL) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: invalid hash level=%u", level); return -1;
}
snprintf(sql, sizeof(sql), "SELECT message_id,node_id,kind,ver,data,signature FROM %s WHERE bucket=? ORDER BY record_key", table);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL) != SQLITE_OK) return fb_db_error(inst, channel, "hash");
sqlite3_bind_int64(st, 1, (sqlite3_int64)(prefix >> 39));
int step, count = 0;
while ((step = sqlite3_step(st)) == SQLITE_ROW) {
struct chat_feedback_record r; uint8_t wire[FB_WIRE_MAX];
if (fb_read_row(st, &r) < 0) { count = -1; break; }
int bytes = fb_encode(&r, wire);
if (bytes < 0 || EVP_DigestUpdate(hash, wire, bytes) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: leaf digest failed ch=%s", channel); count = -1; break;
}
count++;
}
sqlite3_finalize(st);
if (count >= 0 && step != SQLITE_DONE) return fb_db_error(inst, channel, "hash step");
return count;
}
/* Вся страница применяется атомарно; события отправляются после её commit. */
static int fb_apply_page(void* ctx, const char* channel, uint64_t peer, const uint8_t* data, size_t len) {
struct UTUN_INSTANCE* inst = ctx; (void)peer;
if (len < 2) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: short page ch=%s", channel); return -1; }
unsigned count = ((unsigned)data[0] << 8) | data[1]; size_t off = 2;
if (count > (len - 2) / FB_FIXED_SIZE) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: invalid page count=%u bytes=%zu", count, len); return -1;
}
for (unsigned i = 0; i < count; i++) {
struct chat_feedback_record r; int bytes = fb_decode(data + off, len - off, &r);
if (bytes < 0) return -1;
off += bytes;
}
if (off != len) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: trailing page bytes=%zu", len - off); return -1; }
if (!count) return 0;
uint8_t* changed = u_calloc(count, 1);
if (!changed) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: page allocation failed"); return -1; }
if (sqlite3_exec(inst->topo_sqlite_db, "SAVEPOINT feedback_update", NULL, NULL, NULL) != SQLITE_OK) {
u_free(changed); return fb_db_error(inst, channel, "page begin");
}
off = 2; int result = 0;
for (unsigned i = 0; i < count; i++) {
struct chat_feedback_record r; int bytes = fb_decode(data + off, len - off, &r);
if (bytes < 0) { result = -1; break; }
int rc = fb_apply_row(inst, channel, &r); if (rc < 0) { result = -1; break; }
changed[i] = rc > 0; off += bytes;
}
result = fb_finish_transaction(inst, channel, result);
if (result == 0) {
off = 2; int any = 0;
for (unsigned i = 0; i < count; i++) {
struct chat_feedback_record r; int bytes = fb_decode(data + off, len - off, &r); off += bytes;
if (changed[i]) { fb_committed(inst, channel, &r); any = 1; }
}
if (any) merkle_sync_changed(inst, MT_DATA_FEEDBACK, channel);
}
u_free(changed); return result;
}
static int fb_peer(void* ctx, const char* channel, uint64_t peer) {
struct UTUN_INSTANCE* inst = ctx; uint8_t key[32];
return fb_author_key(inst, channel, peer, key) == 0;
}
static const struct merkle_sync_data_ops fb_ops = {
.update_bucket_hash = fb_hash, .get_page = fb_page, .apply_items = fb_apply_page, .validate_peer = fb_peer
};
/* Перестраивает индекс после регистрации модели; подписи проверяются повторно. */
int chat_feedback_init(struct UTUN_INSTANCE* inst) {
if (merkle_sync_register(inst, MT_DATA_FEEDBACK, &fb_ops, inst) < 0) return -1;
sqlite3_stmt* channels = NULL;
if (sqlite3_prepare_v2(inst->topo_sqlite_db, "SELECT channel_id FROM channels", -1, &channels, NULL) != SQLITE_OK)
return fb_db_error(inst, "-", "init channels");
int step, result = 0, records = 0;
while ((step = sqlite3_step(channels)) == SQLITE_ROW) {
char channel[21]; snprintf(channel, sizeof(channel), "%s", sqlite3_column_text(channels, 0));
char table[40], sql[256];
if (chat_feedback_ensure_channel(inst, channel) < 0 || fb_table(inst, channel, table, NULL) < 0) { result = -1; break; }
snprintf(sql, sizeof(sql), "SELECT message_id,node_id,kind,ver,data,signature FROM %s", table);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL) != SQLITE_OK) {
result = fb_db_error(inst, channel, "init records"); break;
}
int rc;
while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
struct chat_feedback_record r; uint64_t id; uint8_t body[FB_WIRE_MAX + 32], key[32]; merkle_key_t record_key;
if (fb_read_row(st, &r) < 0 || fb_table(inst, channel, table, &id) < 0
|| fb_author_key(inst, channel, r.node_id, key) < 0) { result = -1; break; }
int bytes = fb_signed_bytes(id, &r, body);
if (bytes < 0 || sc_ed25519_verify(key, body, bytes, r.signature) != SC_OK || fb_key(id, &r, &record_key) < 0
|| merkle_sync_recompute_path(inst, MT_DATA_FEEDBACK, channel, fb_read64(record_key.bytes)) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: rebuild failed ch=%s node=%016llx",
channel, (unsigned long long)r.node_id); result = -1; break;
}
records++;
}
sqlite3_finalize(st);
if (result < 0) break;
if (rc != SQLITE_DONE) { result = fb_db_error(inst, channel, "init records step"); break; }
}
sqlite3_finalize(channels);
if (result == 0 && step != SQLITE_DONE) return fb_db_error(inst, "-", "init channels step");
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "feedback: initialization result=%d records=%d", result, records); return result;
}
Write Preview
Loading…
Cancel
Save