From 641a8318a87ba96e1bdd90a6ab509fa8253c4b30 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 6 Oct 2026 20:05:18 +0200 Subject: [PATCH] Add signed versioned feedback storage and Merkle adapter --- src/chat/chat_feedback.c | 425 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 425 insertions(+) create mode 100644 src/chat/chat_feedback.c diff --git a/src/chat/chat_feedback.c b/src/chat/chat_feedback.c new file mode 100644 index 00000000..37182e74 --- /dev/null +++ b/src/chat/chat_feedback.c @@ -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 +#include +#include +#include + +#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; +}