media: Ed25519-подписи блоков, local_attrs, MediaBlockReader, файловые вложения
This commit is contained in:
1 parent
6b765de5cc
commit
d4a396f888
20 files changed
+851
-70
No files matched your search
@@ -88,6 +88,7 @@ void si_register(struct DB_SYNC_INSTANCE* si, const char* ch_id) {
|
||||
int chat_core_init(struct UTUN_INSTANCE* inst, const char* db_path) {
|
||||
if (!inst || !db_path) return -1;
|
||||
memset(&g_cc, 0, sizeof(g_cc));
|
||||
snprintf(g_cc.db_path, sizeof(g_cc.db_path), "%s", db_path);
|
||||
|
||||
if (inst->topo_sqlite_db) {
|
||||
g_cc.db = inst->topo_sqlite_db; g_cc.shared_db = 1;
|
||||
|
||||
@@ -28,6 +28,11 @@ int chat_core_is_initialized(void);
|
||||
struct chat_msg_submit {
|
||||
char channel_id[64];
|
||||
char content_type[32];
|
||||
char media_dt[32];
|
||||
char media_basename[256];
|
||||
char media_suffix[8];
|
||||
char media_ext[16];
|
||||
uint32_t media_num_blocks;
|
||||
uint8_t* data;
|
||||
uint32_t data_len;
|
||||
uint64_t timestamp;
|
||||
@@ -80,6 +85,11 @@ void chat_core_connect_channel_trampoline(void* arg);
|
||||
/* Трамплин для gui_bridge_post_uasync (GUI → uasync) */
|
||||
void chat_core_submit_trampoline(void* arg);
|
||||
|
||||
/* Обновление local_attrs для записи (вызывается из uasync-потока) */
|
||||
int chat_core_update_local_attrs(const char* ch_id, uint64_t ts,
|
||||
const uint8_t* author_sig,
|
||||
const char* local_attrs_json);
|
||||
|
||||
/* Сбор статуса (NTP + connections) и отправка в GUI */
|
||||
void chat_core_collect_status_trampoline(void* arg);
|
||||
|
||||
|
||||
@@ -24,6 +24,7 @@
|
||||
struct chat_core_ctx {
|
||||
struct UTUN_INSTANCE* inst;
|
||||
sqlite3* db;
|
||||
char db_path[512];
|
||||
uint8_t shared_db;
|
||||
uint64_t my_node_id;
|
||||
struct DB_SYNC_INSTANCE** si;
|
||||
|
||||
+197
-2
@@ -13,6 +13,15 @@
|
||||
#include "../../lib/platform_compat.h"
|
||||
|
||||
#include <openssl/sha.h>
|
||||
#include <unistd.h>
|
||||
#include <sys/stat.h>
|
||||
|
||||
#define MEDIA_BLOCK_MIN (10 * 1024 * 1024)
|
||||
#define MEDIA_BLOCK_MAX (25 * 1024 * 1024)
|
||||
#define MEDIA_BLOCK_TARGET 30
|
||||
|
||||
/* forward decl */
|
||||
static void chat_core_submit_media_message(struct chat_msg_submit* req);
|
||||
|
||||
/* ─── отправка сообщения (GUI → uasync) ─── */
|
||||
|
||||
@@ -54,14 +63,200 @@ void chat_core_submit_message(struct chat_msg_submit* req) {
|
||||
return;
|
||||
}
|
||||
|
||||
int ret = db_sync_insert_signed(si, json, strlen(json), sig, 64, ts);
|
||||
int ret = db_sync_insert_signed(si, json, strlen(json), sig, 64, ts, NULL);
|
||||
if (ret != 0) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: db_sync_insert_signed failed ret=%d", CC_ID, ret);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
void chat_core_submit_trampoline(void* arg) { chat_core_submit_message((struct chat_msg_submit*)arg); u_free(arg); }
|
||||
void chat_core_submit_trampoline(void* arg) {
|
||||
struct chat_msg_submit* req = (struct chat_msg_submit*)arg;
|
||||
if (req->media_num_blocks > 0)
|
||||
chat_core_submit_media_message(req);
|
||||
else
|
||||
chat_core_submit_message(req);
|
||||
u_free(req);
|
||||
}
|
||||
|
||||
/* ─── Медиа-сообщения: подпись блоков + сборка в единый файл ─── */
|
||||
|
||||
static uint32_t media_calc_block_size(uint64_t file_size) {
|
||||
if (file_size == 0) return 0;
|
||||
uint64_t target = (file_size + MEDIA_BLOCK_TARGET - 1) / MEDIA_BLOCK_TARGET;
|
||||
if (target < MEDIA_BLOCK_MIN) target = MEDIA_BLOCK_MIN;
|
||||
if (target > MEDIA_BLOCK_MAX) target = MEDIA_BLOCK_MAX;
|
||||
return (uint32_t)target;
|
||||
}
|
||||
|
||||
static void chat_core_submit_media_message(struct chat_msg_submit* req) {
|
||||
if (!g_cc.initialized) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: media submit before init", CC_ID); return; }
|
||||
if (!req || req->media_num_blocks == 0) return;
|
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: MEDIA SUBMIT ch=%s ct=%s dt=%s stem=%s ext=%s blocks=%u",
|
||||
CC_ID, req->channel_id, req->content_type,
|
||||
req->media_dt, req->media_basename, req->media_ext, req->media_num_blocks);
|
||||
|
||||
struct DB_SYNC_INSTANCE* si = si_find(req->channel_id);
|
||||
if (!si) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: no db_sync instance for ch=%s", CC_ID, req->channel_id); return; }
|
||||
|
||||
char media_dir[1024];
|
||||
snprintf(media_dir, sizeof(media_dir), "%s/media/%s", g_cc.db_path, req->channel_id);
|
||||
|
||||
uint32_t num_blocks = req->media_num_blocks;
|
||||
uint64_t block_size = 0;
|
||||
uint64_t file_size = 0;
|
||||
|
||||
/* allocate sig array */
|
||||
uint8_t (*sigs)[64] = (uint8_t(*)[64])u_malloc(num_blocks * 64);
|
||||
if (!sigs) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: sig alloc failed blocks=%u", CC_ID, num_blocks); return; }
|
||||
|
||||
/* read blocks, sign, compute sizes */
|
||||
for (uint32_t n = 0; n < num_blocks; n++) {
|
||||
char path[1280];
|
||||
snprintf(path, sizeof(path), "%s/%s_%u_%s_%s.%s",
|
||||
media_dir, req->media_dt, n, req->media_basename, req->media_suffix, req->media_ext);
|
||||
|
||||
FILE* f = fopen(path, "rb");
|
||||
if (!f) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: cannot open block %u: %s", CC_ID, n, path);
|
||||
u_free(sigs); return;
|
||||
}
|
||||
fseek(f, 0, SEEK_END); long fsz = ftell(f); fseek(f, 0, SEEK_SET);
|
||||
if (fsz <= 0) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: empty block %u: %s", CC_ID, n, path); fclose(f); u_free(sigs); return; }
|
||||
|
||||
uint8_t* buf = (uint8_t*)u_malloc((size_t)fsz);
|
||||
if (!buf || fread(buf, 1, (size_t)fsz, f) != (size_t)fsz) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: read fail block %u", CC_ID, n);
|
||||
if (buf) u_free(buf); fclose(f); u_free(sigs); return;
|
||||
}
|
||||
fclose(f);
|
||||
|
||||
if (sc_ed25519_sign(g_cc.inst->my_ed25519_privkey, buf, (size_t)fsz, sigs[n]) != SC_OK) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: Ed25519 sign fail block %u", CC_ID, n);
|
||||
u_free(buf); u_free(sigs); return;
|
||||
}
|
||||
u_free(buf);
|
||||
|
||||
file_size += (uint64_t)fsz;
|
||||
if (n == 0) block_size = (uint64_t)fsz;
|
||||
}
|
||||
|
||||
if (file_size == 0) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: zero file size", CC_ID); u_free(sigs); return; }
|
||||
|
||||
/* build final data: <base>|<file_size>|<block_size>|<num_blocks>|<sig_hex>,... */
|
||||
size_t base_len = req->data_len;
|
||||
size_t sigs_hex_len = num_blocks * 129; /* 128 hex + comma */
|
||||
size_t full_data_cap = base_len + 128 + sigs_hex_len;
|
||||
char* full_data = (char*)u_malloc(full_data_cap + 1);
|
||||
if (!full_data) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: data alloc fail", CC_ID); u_free(sigs); return; }
|
||||
|
||||
size_t off = 0;
|
||||
if (base_len > 0) { memcpy(full_data + off, req->data, base_len); off += base_len; }
|
||||
off += snprintf(full_data + off, full_data_cap - off, "|%llu|%llu|%u|",
|
||||
(unsigned long long)file_size, (unsigned long long)block_size, num_blocks);
|
||||
for (uint32_t n = 0; n < num_blocks; n++) {
|
||||
if (n > 0) full_data[off++] = ',';
|
||||
for (int b = 0; b < 64; b++)
|
||||
off += snprintf(full_data + off, full_data_cap - off, "%02x", sigs[n][b]);
|
||||
}
|
||||
size_t full_data_len = off;
|
||||
|
||||
/* build JSON with dynamic buffer */
|
||||
size_t json_cap = full_data_len + 256;
|
||||
char* json = (char*)u_malloc(json_cap);
|
||||
if (!json) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: json alloc fail", CC_ID); u_free(full_data); u_free(sigs); return; }
|
||||
snprintf(json, json_cap,
|
||||
"{\"n\":%llu,\"ch\":\"%s\",\"ct\":\"%s\",\"d\":\"%.*s\"}",
|
||||
(unsigned long long)g_cc.my_node_id, req->channel_id,
|
||||
req->content_type, (int)full_data_len, full_data);
|
||||
|
||||
uint64_t ts = db_sync_next_timestamp(si);
|
||||
size_t sig_msg_len = 8 + strlen(json);
|
||||
uint8_t* sig_msg = (uint8_t*)u_malloc(sig_msg_len);
|
||||
if (!sig_msg) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: sig_msg alloc fail", CC_ID); u_free(json); u_free(full_data); u_free(sigs); return; }
|
||||
memcpy(sig_msg, &ts, 8);
|
||||
memcpy(sig_msg + 8, json, sig_msg_len - 8);
|
||||
uint8_t json_sig[64];
|
||||
if (sc_ed25519_sign(g_cc.inst->my_ed25519_privkey, sig_msg, sig_msg_len, json_sig) != SC_OK) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: Ed25519 sign JSON failed", CC_ID);
|
||||
u_free(sig_msg); u_free(json); u_free(full_data); u_free(sigs); return;
|
||||
}
|
||||
|
||||
/* assemble single file from blocks, then delete blocks */
|
||||
{
|
||||
char assembled_path[1280];
|
||||
snprintf(assembled_path, sizeof(assembled_path), "%s/%s_%s_%s.%s",
|
||||
media_dir, req->media_dt, req->media_basename, req->media_suffix, req->media_ext);
|
||||
FILE* af = fopen(assembled_path, "wb");
|
||||
if (af) {
|
||||
for (uint32_t n = 0; n < num_blocks; n++) {
|
||||
char block_path[1280];
|
||||
snprintf(block_path, sizeof(block_path), "%s/%s_%u_%s_%s.%s",
|
||||
media_dir, req->media_dt, n, req->media_basename, req->media_suffix, req->media_ext);
|
||||
FILE* bf = fopen(block_path, "rb");
|
||||
if (bf) {
|
||||
uint8_t buf[65536]; size_t rd;
|
||||
while ((rd = fread(buf, 1, sizeof(buf), bf)) > 0) fwrite(buf, 1, rd, af);
|
||||
fclose(bf);
|
||||
unlink(block_path);
|
||||
}
|
||||
}
|
||||
fclose(af);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: assembled %s (%llu bytes)", CC_ID, assembled_path, (unsigned long long)file_size);
|
||||
} else {
|
||||
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: cannot create assembled file %s", CC_ID, assembled_path);
|
||||
}
|
||||
}
|
||||
|
||||
/* build local_attrs JSON */
|
||||
char local_attrs_json[512];
|
||||
{
|
||||
char bl_list[256] = "0"; size_t bl_off = 1;
|
||||
for (uint32_t n = 1; n < num_blocks; n++)
|
||||
bl_off += snprintf(bl_list + bl_off, sizeof(bl_list) - bl_off, ",%u", n);
|
||||
snprintf(local_attrs_json, sizeof(local_attrs_json),
|
||||
"{\"st\":\"fl\",\"fp\":\"%s_%s_%s.%s\",\"bl\":[%s]}",
|
||||
req->media_dt, req->media_basename, req->media_suffix, req->media_ext, bl_list);
|
||||
}
|
||||
|
||||
int ret = db_sync_insert_signed(si, json, strlen(json), json_sig, 64, ts, local_attrs_json);
|
||||
if (ret != 0)
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: media insert failed ret=%d", CC_ID, ret);
|
||||
|
||||
u_free(sig_msg); u_free(json); u_free(full_data); u_free(sigs);
|
||||
}
|
||||
|
||||
/* ─── Обновление local_attrs (для будущей приёмной стороны) ─── */
|
||||
|
||||
int chat_core_update_local_attrs(const char* ch_id, uint64_t ts,
|
||||
const uint8_t* author_sig,
|
||||
const char* local_attrs_json)
|
||||
{
|
||||
if (!g_cc.initialized || !ch_id || !author_sig || !local_attrs_json) return -1;
|
||||
|
||||
char tbl[80]; msg_table_name(ch_id, tbl, sizeof(tbl));
|
||||
char sql[256];
|
||||
snprintf(sql, sizeof(sql),
|
||||
"UPDATE \"%s\" SET local_attrs=? WHERE timestamp=? AND author_signature=?", tbl);
|
||||
|
||||
sqlite3_stmt* stmt = NULL;
|
||||
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &stmt, NULL) != SQLITE_OK) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: update local_attrs prep fail: %s", CC_ID, sqlite3_errmsg(g_cc.db));
|
||||
return -1;
|
||||
}
|
||||
sqlite3_bind_text(stmt, 1, local_attrs_json, -1, SQLITE_STATIC);
|
||||
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)ts);
|
||||
sqlite3_bind_blob(stmt, 3, author_sig, DB_SIG_SIZE, SQLITE_STATIC);
|
||||
int rc = sqlite3_step(stmt);
|
||||
sqlite3_finalize(stmt);
|
||||
if (rc != SQLITE_DONE) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: update local_attrs step fail rc=%d", CC_ID, rc);
|
||||
return -1;
|
||||
}
|
||||
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: local_attrs updated ch=%s ts=%llu", CC_ID, ch_id, (unsigned long long)ts);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* ─── DB-операции для chat_sync ─── */
|
||||
|
||||
|
||||
+20
-10
@@ -15,6 +15,7 @@
|
||||
#include "../../lib/platform_compat.h"
|
||||
#include <string.h>
|
||||
#include <openssl/evp.h>
|
||||
#include <unistd.h>
|
||||
|
||||
// ---- Forward declarations ----
|
||||
struct DB_SYNC;
|
||||
@@ -422,7 +423,8 @@ static int db_verify_author_sig(struct DB_SYNC_INSTANCE* si,
|
||||
static int db_record_insert(struct DB_SYNC_INSTANCE* si,
|
||||
uint64_t id, uint64_t ts, uint64_t author_node_id,
|
||||
const char* json, size_t jlen,
|
||||
const uint8_t* author_sig, int do_cascade)
|
||||
const uint8_t* author_sig, int do_cascade,
|
||||
const char* local_attrs)
|
||||
{
|
||||
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "id=%llu ts=%llu author=%N len=%zu cascade=%d", (unsigned long long)id, (unsigned long long)ts, (unsigned long long)author_node_id, jlen, do_cascade);
|
||||
sqlite3* db = SI_DB(si);
|
||||
@@ -465,8 +467,8 @@ static int db_record_insert(struct DB_SYNC_INSTANCE* si,
|
||||
if (si_prep(si, &stmt,
|
||||
"INSERT INTO \"%s\""
|
||||
" (timestamp,node_id,id,chain_hash,flags,data,"
|
||||
" author_signature,delivered_peers,delivery_chain)"
|
||||
" VALUES (?,?,?,?,0,?,?,0,'')") != SQLITE_OK)
|
||||
" author_signature,local_attrs,delivered_peers,delivery_chain)"
|
||||
" VALUES (?,?,?,?,0,?,?,?,0,'')") != SQLITE_OK)
|
||||
{ DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_record_insert INSERT prep: %s", sqlite3_errmsg(db)); sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL); return -1; }
|
||||
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)ts);
|
||||
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)author_node_id);
|
||||
@@ -475,6 +477,8 @@ static int db_record_insert(struct DB_SYNC_INSTANCE* si,
|
||||
if (jlen > 0) sqlite3_bind_blob(stmt, 5, json, (int)jlen, SQLITE_STATIC);
|
||||
else sqlite3_bind_null(stmt, 5);
|
||||
sqlite3_bind_blob(stmt, 6, author_sig, DB_SIG_SIZE, SQLITE_STATIC);
|
||||
if (local_attrs && local_attrs[0]) sqlite3_bind_text(stmt, 7, local_attrs, -1, SQLITE_STATIC);
|
||||
else sqlite3_bind_text(stmt, 7, "", -1, SQLITE_STATIC);
|
||||
rc = sqlite3_step(stmt);
|
||||
sqlite3_finalize(stmt);
|
||||
if (rc != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "INSERT: %s", sqlite3_errmsg(db)); sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL); return -1; }
|
||||
@@ -923,7 +927,7 @@ static void db_handle_send_data(struct DB_SYNC_INSTANCE* si, uint64_t src, const
|
||||
const uint8_t* save = ptr;
|
||||
uint64_t rid, rts, rauthor; uint32_t rdlen; const uint8_t* rdata, *rsig; int rsiglen;
|
||||
if (si_parse_record(&ptr, p + len, &rid, &rts, &rauthor, &rdlen, &rdata, &rsig, &rsiglen) != 0) { parse_fails++; break; }
|
||||
int ret = db_record_insert(si, rid, rts, rauthor, (const char*)rdata, rdlen, rsig, 0);
|
||||
int ret = db_record_insert(si, rid, rts, rauthor, (const char*)rdata, rdlen, rsig, 0, NULL);
|
||||
if (ret >= 0) { recv_ts[recv_cnt] = rts; if (rsig && rsiglen >= 8) memcpy(&recv_sig8[recv_cnt], rsig, 8); else recv_sig8[recv_cnt] = 0; recv_cnt++; }
|
||||
if (ret >= 0) { if (received == 0) first_id = rid; last_id = rid; received++; }
|
||||
if (ret == 1) duplicates++;
|
||||
@@ -1160,7 +1164,7 @@ static void db_handle_push(struct DB_SYNC_INSTANCE* si, uint64_t src, const uint
|
||||
if (si_parse_record(&ptr, p + len, &rid, &rts, &rauthor, &rdlen, &rdata, &rsig, &rsiglen) != 0) return;
|
||||
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "from=%016llx id=%llu author=%016llx ts=%llu len=%u", (unsigned long long)src, (unsigned long long)rid, (unsigned long long)rauthor, (unsigned long long)rts, rdlen);
|
||||
|
||||
int ret = db_record_insert(si, rid, rts, rauthor, (const char*)rdata, rdlen, rsig, 0);
|
||||
int ret = db_record_insert(si, rid, rts, rauthor, (const char*)rdata, rdlen, rsig, 0, NULL);
|
||||
if (ret == 0) {
|
||||
uint32_t ins_pos = si_find_pos(si, rts, rsig);
|
||||
uint32_t total = db_count(si);
|
||||
@@ -1672,6 +1676,7 @@ struct DB_SYNC_INSTANCE* db_sync_instance_add(struct UTUN_INSTANCE* inst, const
|
||||
" flags INTEGER NOT NULL DEFAULT 0,"
|
||||
" data BLOB,"
|
||||
" author_signature BLOB NOT NULL,"
|
||||
" local_attrs TEXT DEFAULT '',"
|
||||
" delivered_peers INTEGER NOT NULL DEFAULT 0,"
|
||||
" delivery_chain TEXT NOT NULL DEFAULT '',"
|
||||
" PRIMARY KEY (timestamp, author_signature))",
|
||||
@@ -1746,7 +1751,8 @@ void db_sync_instance_remove(struct DB_SYNC_INSTANCE* si)
|
||||
}
|
||||
|
||||
int db_sync_insert_signed(struct DB_SYNC_INSTANCE* si, const char* json_data, size_t len,
|
||||
const uint8_t* sig, size_t sig_len, uint64_t ts)
|
||||
const uint8_t* sig, size_t sig_len, uint64_t ts,
|
||||
const char* local_attrs)
|
||||
{
|
||||
if (!si || !si->enabled || !json_data || len == 0) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "insert invalid args"); return -1; }
|
||||
if (!sig || sig_len != DB_SIG_SIZE) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "insert_signed: signature required (64 bytes Ed25519)"); return -1; }
|
||||
@@ -1755,22 +1761,25 @@ int db_sync_insert_signed(struct DB_SYNC_INSTANCE* si, const char* json_data, si
|
||||
uint64_t id = si->next_id;
|
||||
uint64_t author_node_id = si->db_sync->inst->node_id;
|
||||
|
||||
int ret = db_record_insert(si, id, ts, author_node_id, json_data, len, sig, 1);
|
||||
int ret = db_record_insert(si, id, ts, author_node_id, json_data, len, sig, 1, local_attrs);
|
||||
if (ret != 0) return ret;
|
||||
if (si->on_insert) si->on_insert(si, ts, json_data, len, author_node_id, si->on_insert_arg);
|
||||
|
||||
// Build PUSH: [DB_MSG_PUSH][id:8][ts:8][author:8][dlen:4][data][sig_len:1=64][sig:64]
|
||||
uint32_t sl = DB_SIG_SIZE;
|
||||
uint8_t pbuf[2560];
|
||||
size_t push_overhead = 1 + 8 + 8 + 8 + 4 + 1 + 64;
|
||||
size_t push_total = push_overhead + (len > 0 ? len : 0);
|
||||
uint8_t* pbuf = (uint8_t*)u_malloc(push_total);
|
||||
if (!pbuf) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "push alloc failed size=%zu", push_total); return -1; }
|
||||
uint32_t off = 0;
|
||||
pbuf[off++] = DB_MSG_PUSH;
|
||||
memcpy(pbuf + off, &id, 8); off += 8;
|
||||
memcpy(pbuf + off, &ts, 8); off += 8;
|
||||
memcpy(pbuf + off, &author_node_id, 8); off += 8;
|
||||
memcpy(pbuf + off, &len, 4); off += 4;
|
||||
if (len > 0 && off + len <= sizeof(pbuf)) { memcpy(pbuf + off, json_data, len); off += len; }
|
||||
if (len > 0) { memcpy(pbuf + off, json_data, len); off += (uint32_t)len; }
|
||||
pbuf[off++] = (uint8_t)sl;
|
||||
if (off + sl <= sizeof(pbuf)) { memcpy(pbuf + off, sig, sl); off += sl; }
|
||||
memcpy(pbuf + off, sig, sl); off += sl;
|
||||
|
||||
// Push to all synced peers
|
||||
int push_count = 0;
|
||||
@@ -1783,6 +1792,7 @@ int db_sync_insert_signed(struct DB_SYNC_INSTANCE* si, const char* json_data, si
|
||||
if (push_count > 0)
|
||||
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "sync [%s:ME--] → PUSH out: new msg id=%llu ts=%llu → forwarded to %d synced peers",
|
||||
SI_SHRT(si), (unsigned long long)id, (unsigned long long)ts, push_count);
|
||||
u_free(pbuf);
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
||||
+2
-1
@@ -96,7 +96,8 @@ void db_sync_instance_remove(struct DB_SYNC_INSTANCE* si);
|
||||
// sig = Ed25519(ts[8] || author[8] || json).
|
||||
// ts — call db_sync_next_timestamp(si) before signing to reserve monotonically increasing timestamp.
|
||||
int db_sync_insert_signed(struct DB_SYNC_INSTANCE* si, const char* json_data, size_t len,
|
||||
const uint8_t* sig, size_t sig_len, uint64_t ts);
|
||||
const uint8_t* sig, size_t sig_len, uint64_t ts,
|
||||
const char* local_attrs);
|
||||
uint32_t db_sync_count(struct DB_SYNC_INSTANCE* si);
|
||||
uint64_t db_sync_get_last_timestamp(struct DB_SYNC_INSTANCE* si);
|
||||
uint64_t db_sync_next_timestamp(struct DB_SYNC_INSTANCE* si);
|
||||
|
||||
Reference in new issue
Block a user