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

feat(chat): add local DM deletion and improve Android member list

Browse Source 
Delete local conversations atomically, retain replay protection and sequence counters, and guard asynchronous media work by history generation. Add Android avatar menu and confirmation, persistent file cleanup, and regression coverage.

Fix unnamed member wire padding and compact Android member action spacing.

Validation: 158 core tests passed; targeted DM tests, desktop build, Android unit tests and debug APK build passed.
This commit is contained in:
evgeny committed 2026-10-18 02:04:05 +02:00
1 parent 97710b609f
commit be295f768c
38 files changed
+1005 -126
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/Makefile.am
+4
View File
@@ -175,6 +175,8 @@ utun_CORE_SOURCES = \
chat/invite_build.c \
chat/chat_headless_control.c \
dm/dm_core.c \
dm/dm_history.c \
dm/dm_history.h \
dm/dm_crypto.c \
dm/dm_mailbox.c \
dm/dm_media.c \
@@ -340,6 +342,8 @@ libutun_a_SOURCES = \
chat/invite_build.c \
chat/chat_headless_control.c \
dm/dm_core.c \
dm/dm_history.c \
dm/dm_history.h \
dm/dm_crypto.c \
dm/dm_mailbox.c \
dm/dm_media.c \
src/chat/chat_event.c
+1
View File
@@ -53,6 +53,7 @@ void chat_event_post(struct UTUN_INSTANCE* inst, int type, const uint8_t* data,
[55]="RADIO_MUTE_RESULT",
[56]="MSG_SUBMITTED",
[57]="CPU_LOAD",
[59]="DM_DELETE_RESULT",
};
const char* n = (type >= 1 && type < (int)(sizeof(names) / sizeof(names[0]))) ? names[type] : "?";
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT, "chat_event: %s(%d) data=%d bytes", n, type, len);
src/chat/chat_event.h
+3 -2
View File
@@ -53,7 +53,7 @@ extern "C" {
#define CHAT_EVT_INVITE_CANDIDATES 30 /* [ch_id_len:1][ch_id:var][auto_node_id:8][count:2][node_id:8]* — достижимые узлы для invite */
#define CHAT_EVT_CHANNEL_DELETED 31 /* [ch_id_len:1][ch_id:var] — канал удалён локально */
#define CHAT_EVT_DM_MSG_RECEIVED 32 /* [conv_id_len:1][conv_id:var][author_node_id:8] */
#define CHAT_EVT_DM_CONV_UPDATED 33 /* [conv_id_len:1][conv_id:var] — беседа создана/изменена */
#define CHAT_EVT_DM_CONV_UPDATED 33 /* [conv_id_len:1][conv_id:var][generation:8LE] — беседа создана/изменена */
#define CHAT_EVT_CALL_INCOMING 34 /* [ch_id_len:1][ch_id:var][caller_node_id:8][call_id:8] */
#define CHAT_EVT_CALL_RINGING 35 /* [call_id:8] */
#define CHAT_EVT_CALL_ACCEPTED 36 /* [call_id:8] */
@@ -69,7 +69,7 @@ extern "C" {
#define CHAT_EVT_ATTACHMENT_PREPARATION 46 /* [target_len:1][target:var][job_uuid:16][state:1], 0=start 1=ready 2=failed */
#define CHAT_EVT_DM_CALL_ROUTE 47 /* [conv_id_len:1][conv_id:var][peer_node_id:8][group_id:8], 0 = нет BGP-пути */
#define CHAT_EVT_DM_INVITE_RESULT 48 /* [request_id:8][result:4][peer_node_id:8][conv_id:8], dm_invite_result */
#define CHAT_EVT_DM_INVITE_RESULT 48 /* [request_id:8][result:4][peer_node_id:8][conv_id:8][generation:8], dm_invite_result */
#define CHAT_EVT_MEMBER_FILES_RESPONSE 49 /* [group:8LE][server:8LE][mf wire response] */
#define CHAT_EVT_MEMBER_FILES_TRANSFER 50 /* id/group/server/owner/done/total:8LE,status:2LE,state:1 */
#define CHAT_EVT_AVATAR_RESULT 51 /* request_id/group:8LE, failed:1; результат публикации ссылки профиля */
@@ -80,6 +80,7 @@ extern "C" {
#define CHAT_EVT_MSG_SUBMITTED 56 /* [request_id:8LE][success:1] — результат локального сохранения сообщения */
#define CHAT_EVT_CPU_LOAD 57 /* [valid:1][average_percent:1] — CPU потока uasync, среднее за секунду, 0..100 */
#define CHAT_EVT_STREAM_SOURCES 58 /* [group_id:8BE][valid:1][JSON:utf8] — доступные публикации выбранного канала */
#define CHAT_EVT_DM_DELETE_RESULT 59 /* [request:8LE][conv:8LE][generation:8LE][success:1] — после транзакции */
typedef void (*chat_event_handler_fn)(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len);
src/chat/chat_member.c
+5 -1
View File
@@ -85,10 +85,14 @@ void chat_core_serialize_member(const struct chat_member_display* src, uint8_t*
memcpy(dst + 11, &rtt, 2);
dst[13] = src->nat_type;
dst[14] = src->storage;
memcpy(dst + 15, src->name, 64);
/* Поле на проводе всегда завершается NUL и не содержит хвост памяти исходной строки. */
strncpy((char*)dst + 15, src->name, sizeof(src->name) - 1);
dst[78] = 0;
mf_put64(dst + 79, src->avatar_server);
memcpy(dst + 87, src->avatar_file, 70);
dst[157] = src->storage_role;
DEBUG_TRACE(DEBUG_CATEGORY_CHAT_SYNC, "member snapshot: serialized node=%016llx name_bytes=%zu zero_padded=1",
(unsigned long long)src->node_id, strlen((const char*)dst + 15));
}
int chat_core_deserialize_member(const uint8_t* src, struct chat_member_display* dst) {
src/dm/dm_core.c
+58 -11
View File
@@ -11,6 +11,7 @@
#include "dm_core.h"
#include "dm_crypto.h"
#include "dm_history.h"
#include "dm_mailbox.h"
#include "../chat/chat_core.h"
#include "../chat/chat_event.h"
@@ -450,12 +451,15 @@ int dm_accept_message(struct UTUN_INSTANCE* inst, const uint8_t* msg, size_t len
return -1;
}
if (dm_verify_message(inst, inst->node_id, msg, len) != 0) return -1;
int deleted = dm_history_deleted(inst, msg, len, 0, NULL);
if (deleted < 0) return -1;
uint64_t conv, seq, author;
memcpy(&conv, msg, 8);
memcpy(&seq, msg + 8, 8);
memcpy(&author, msg + 24, 8);
char conv_id[64];
snprintf(conv_id, sizeof(conv_id), "%llu", (unsigned long long)conv);
if (deleted) goto receipt;
struct dm_conv c;
int created = dm_conv_load(dm, conv_id, &c) != 0;
if (created) {
@@ -490,10 +494,12 @@ int dm_accept_message(struct UTUN_INSTANCE* inst, const uint8_t* msg, size_t len
uint64_t ts;
memcpy(&ts, msg + 16, 8);
if (ts > c.last_ts) c.last_ts = ts;
if (inserted < 0 || dm_conv_save(dm, &c) != 0 || dm_media_track(inst, 0, msg, len) != 0 || dm_exec(dm, "COMMIT") != 0) {
if (inserted < 0 || dm_conv_save(dm, &c) != 0 || dm_media_track(inst, 0, msg, len) != 0 ||
(inserted && dm_history_show(inst, conv_id)) || dm_exec(dm, "COMMIT") != 0) {
dm_exec(dm, "ROLLBACK");
return -1;
}
receipt:
memcpy(receipt, "DMACK001", 8);
memcpy(receipt + 8, msg, 16);
memcpy(receipt + 24, &author, 8);
@@ -503,6 +509,7 @@ int dm_accept_message(struct UTUN_INSTANCE* inst, const uint8_t* msg, size_t len
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: receipt signing failed conv=%s seq=%llu", DM_ID, conv_id, (unsigned long long)seq);
return -1;
}
if (deleted) return 0;
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: committed receive conv=%s seq=%llu new=%d plain=%zu", DM_ID,
conv_id, (unsigned long long)seq, inserted, plain_len);
if (inserted) {
@@ -511,7 +518,12 @@ int dm_accept_message(struct UTUN_INSTANCE* inst, const uint8_t* msg, size_t len
evt[0] = (uint8_t)cl;
memcpy(evt + 1, conv_id, cl);
memcpy(evt + 1 + cl, &author, 8);
if (created) chat_event_post(inst, CHAT_EVT_DM_CONV_UPDATED, evt, 1 + (int)cl);
if (created) {
uint64_t generation = dm_history_generation(inst, conv_id);
memcpy(evt + 1 + cl, &generation, 8);
chat_event_post(inst, CHAT_EVT_DM_CONV_UPDATED, evt, 1 + (int)cl + 8);
memcpy(evt + 1 + cl, &author, 8);
}
chat_event_post(inst, CHAT_EVT_DM_MSG_RECEIVED, evt, 1 + (int)cl + 8);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: received conv=%s seq=%llu author=%llu", DM_ID, conv_id,
(unsigned long long)seq, (unsigned long long)author);
@@ -739,13 +751,17 @@ int dm_start(struct UTUN_INSTANCE* inst, uint64_t peer_node_id,
(unsigned long long)c.group_id);
}
if (dm_history_show(inst, conv_id)) return -1;
/* событие GUI (всегда, чтобы GUI мог открыть/выделить беседу) */
{
size_t cl = strlen(conv_id);
uint8_t evt[1 + 64];
uint8_t evt[1 + 64 + 8];
evt[0] = (uint8_t)cl;
memcpy(evt + 1, conv_id, cl);
chat_event_post(inst, CHAT_EVT_DM_CONV_UPDATED, evt, 1 + (int)cl);
uint64_t generation = dm_history_generation(inst, conv_id);
memcpy(evt + 1 + cl, &generation, 8);
chat_event_post(inst, CHAT_EVT_DM_CONV_UPDATED, evt, 1 + (int)cl + 8);
}
return 0;
}
@@ -763,6 +779,9 @@ int dm_send(struct UTUN_INSTANCE* inst, const char* conv_id, const char* content
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: send conversation missing/seq exhausted conv=%s", DM_ID, conv_id);
return -1;
}
if (!dm_history_current(inst, conv_id, dm_history_generation(inst, conv_id))) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM send: hidden conversation conv=%s", conv_id); return -1;
}
uint64_t seq = c.last_out_seq + 1;
uint64_t ts = (uint64_t)(ntp_time_get_us(inst) / 1000);
uint8_t* body = NULL;
@@ -815,7 +834,7 @@ int dm_core_init(struct UTUN_INSTANCE* inst) {
inst->dm = dm;
dm->inst = inst;
dm->db = inst->topo_sqlite_db;
if (!dm->db || dm_create_tables(dm) != 0) goto fail;
if (!dm->db || dm_create_tables(dm) != 0 || dm_history_init(inst)) goto fail;
if (etcp_router_bind(inst, ETCP_RT_ID_DM, dm_recv_cb) != 0) goto fail;
dm->initialized = 1;
dm->timer = uasync_set_timeout(inst->ua, 10000, dm, dm_tick, "dm_outbox");
@@ -877,8 +896,9 @@ int dm_list_conversations_json(struct UTUN_INSTANCE* inst, char* buf, size_t buf
int failed = strbuf_addf(&json, "[") < 0;
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db,
"SELECT conv_id, peer_node_id, COALESCE(peer_name,''), last_ts, last_out_seq, last_in_seq"
" FROM dm_conversations ORDER BY last_ts DESC LIMIT 200",
"SELECT c.conv_id,peer_node_id,COALESCE(peer_name,''),last_ts,last_out_seq,last_in_seq,COALESCE(h.generation,1)"
" FROM dm_conversations c LEFT JOIN dm_history h USING(conv_id) WHERE COALESCE(h.hidden,0)=0"
" ORDER BY last_ts DESC LIMIT 200",
-1, &st, NULL) == SQLITE_OK) {
int first = 1;
while (sqlite3_step(st) == SQLITE_ROW) {
@@ -898,10 +918,11 @@ int dm_list_conversations_json(struct UTUN_INSTANCE* inst, char* buf, size_t buf
json_escape(name ? name : "", esc_name, sizeof(esc_name));
if (strbuf_addf(&json,
"%s{\"conv_id\":\"%s\",\"peer_node_id\":%llu,\"peer_name\":\"%s\","
"\"last_ts\":%llu,\"out_seq\":%llu,\"in_seq\":%llu}",
"\"last_ts\":%llu,\"out_seq\":%llu,\"in_seq\":%llu,\"generation\":%llu}",
first ? "" : ",", conv_id, (unsigned long long)peer,
esc_name, (unsigned long long)last_ts,
(unsigned long long)out_seq, (unsigned long long)in_seq) < 0) { failed = 1; break; }
(unsigned long long)out_seq, (unsigned long long)in_seq,
(unsigned long long)sqlite3_column_int64(st, 6)) < 0) { failed = 1; break; }
first = 0;
}
if (sqlite3_errcode(dm->db) != SQLITE_DONE && sqlite3_errcode(dm->db) != SQLITE_OK) failed = 1;
@@ -1045,7 +1066,8 @@ void dm_send_trampoline(void* arg) {
if (!req) return;
struct UTUN_INSTANCE* inst = req->inst;
char conv[64]; strncpy(conv, req->conv_id, sizeof(conv) - 1); conv[sizeof(conv) - 1] = '\0';
int rc = dm_send(inst, conv, "text", req->data, req->data_len);
uint64_t generation = req->generation ? req->generation : dm_history_generation(inst, conv);
int rc = dm_history_current(inst, conv, generation) ? dm_send(inst, conv, "text", req->data, req->data_len) : -1;
u_free(arg);
if (rc != 0)
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: dm_send_trampoline — dm_send failed conv=%s",
@@ -1058,9 +1080,11 @@ static void dm_invite_done(struct dm_invite_req* req, int result) {
if (dm && dm->invite == req) dm->invite = NULL;
uint64_t conv = dm_derive_conv_id(req->inst->node_id, req->peer_node_id);
int32_t status = result;
uint8_t event[28];
uint8_t event[36];
memcpy(event, &req->request_id, 8); memcpy(event + 8, &status, 4);
memcpy(event + 12, &req->peer_node_id, 8); memcpy(event + 20, &conv, 8);
char id[64]; snprintf(id, sizeof(id), "%llu", (unsigned long long)conv);
uint64_t generation = dm_history_generation(req->inst, id); memcpy(event + 28, &generation, 8);
DEBUG_INFO(DEBUG_CATEGORY_DM, "invite PM: finished request=%llu peer=%llu ch=%llu result=%d",
(unsigned long long)req->request_id, (unsigned long long)req->peer_node_id,
(unsigned long long)req->channel_id, result);
@@ -1080,6 +1104,7 @@ static void dm_invite_ready(void* arg, int result, const char* link) {
}
char conv[64];
snprintf(conv, sizeof(conv), "%llu", (unsigned long long)dm_derive_conv_id(req->inst->node_id, req->peer_node_id));
if (!dm_history_current(req->inst, conv, req->generation)) { dm_invite_done(req, DM_INVITE_CANCELLED); return; }
size_t bytes = strlen(link);
sqlite3_stmt* st = NULL;
char channel[64], payload[DM_DATA_MAX];
@@ -1147,6 +1172,9 @@ void dm_invite_trampoline(void* arg) {
dm_invite_done(req, DM_INVITE_START_ERROR); return;
}
}
if (dm_history_show(req->inst, conv)) { dm_invite_done(req, DM_INVITE_START_ERROR); return; }
req->generation = dm_history_generation(req->inst, conv);
if (!dm_history_current(req->inst, conv, req->generation)) { dm_invite_done(req, DM_INVITE_CANCELLED); return; }
DEBUG_INFO(DEBUG_CATEGORY_DM, "invite PM: building request=%llu peer=%llu source=%s ch=%s conv=%s",
(unsigned long long)req->request_id, (unsigned long long)req->peer_node_id, req->source_ch_id, channel, conv);
dm->invite = req;
@@ -1154,6 +1182,25 @@ void dm_invite_trampoline(void* arg) {
if (!req->pending) dm_invite_ready(req, CHAT_JOIN_ERROR, NULL);
}
void dm_cancel_pending_history(struct UTUN_INSTANCE* inst, const char* conv) {
struct dm_state* dm = dm_of(inst);
if (dm && dm->invite && dm_derive_conv_id(inst->node_id, dm->invite->peer_node_id) == strtoull(conv, NULL, 10))
chat_invite_build_cancel(dm->invite->pending);
}
void dm_delete_trampoline(void* arg) {
struct dm_delete_req* req = arg;
if (!req) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM delete: missing request"); return; }
int success = dm_history_delete(req->inst, req->conv_id, req->generation) == 0;
uint64_t conv = strtoull(req->conv_id, NULL, 10);
uint8_t event[25];
memcpy(event, &req->request_id, 8); memcpy(event + 8, &conv, 8); memcpy(event + 16, &req->generation, 8); event[24] = success;
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM delete result: request=%llu conv=%s generation=%llu success=%d",
(unsigned long long)req->request_id, req->conv_id, (unsigned long long)req->generation, success);
chat_event_post(req->inst, CHAT_EVT_DM_DELETE_RESULT, event, sizeof(event));
u_free(req);
}
/* Единая BGP-проверка для кнопки PM и запуска звонка; только uasync-поток. */
int dm_call_route(struct UTUN_INSTANCE* inst, const char* conv_id, uint64_t* peer, uint64_t* group_id) {
struct dm_state* dm = dm_of(inst);
src/dm/dm_core.h
+7 -1
View File
@@ -86,6 +86,11 @@ int dm_start(struct UTUN_INSTANCE* inst, uint64_t peer_node_id,
int dm_send(struct UTUN_INSTANCE* inst, const char* conv_id, const char* content_type,
const uint8_t* data, uint32_t data_len);
/* Локальное удаление; generation защищает повторно открытую беседу от старого запроса. */
struct dm_delete_req { struct UTUN_INSTANCE* inst; char conv_id[64]; uint64_t generation, request_id; };
void dm_delete_trampoline(void* arg);
void dm_cancel_pending_history(struct UTUN_INSTANCE* inst, const char* conv);
/* Список бесед в JSON: [{conv_id,peer_node_id,peer_name,last_ts,out_seq,in_seq}] */
int dm_list_conversations_json(struct UTUN_INSTANCE* inst, char* buf, size_t buf_size, size_t* out_len);
@@ -111,6 +116,7 @@ void dm_start_trampoline(void* arg);
struct dm_send_req {
struct UTUN_INSTANCE* inst;
char conv_id[64];
uint64_t generation; /* 0: захватить при входе в uasync, до любой фоновой работы. */
uint32_t data_len;
uint8_t* data; /* data_len байт (выделяется в хвосте структуры) */
};
@@ -127,7 +133,7 @@ enum dm_invite_result {
};
struct dm_invite_req {
struct UTUN_INSTANCE* inst;
uint64_t request_id, peer_node_id, channel_id;
uint64_t request_id, peer_node_id, channel_id, generation;
char source_ch_id[64]; /* общая группа для новой PM; из шапки PM может быть пустой */
char peer_name[128];
int cancel;
src/dm/dm_history.c
+177
View File
@@ -0,0 +1,177 @@
#include "dm_history.h"
#include "dm_core.h"
#include "dm_crypto.h"
#include "../utun_instance.h"
#include "../../lib/debug_config.h"
#include <sqlite3.h>
#include <openssl/sha.h>
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
static int history_error(struct UTUN_INSTANCE* inst, const char* operation, const char* conv) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM history: %s conv=%s sql=%s", operation, conv ? conv : "",
inst && inst->topo_sqlite_db ? sqlite3_errmsg(inst->topo_sqlite_db) : "database unavailable");
return -1;
}
int dm_history_init(struct UTUN_INSTANCE* inst) {
const char* sql =
"CREATE TABLE IF NOT EXISTS dm_history(conv_id TEXT PRIMARY KEY,generation INTEGER NOT NULL,hidden INTEGER NOT NULL);"
"CREATE TABLE IF NOT EXISTS dm_deleted_messages(conv_id TEXT NOT NULL,dir INTEGER NOT NULL,seq INTEGER NOT NULL,"
"body_hash BLOB NOT NULL,media_id BLOB,media_receipt BLOB,PRIMARY KEY(conv_id,dir,seq));"
"CREATE TABLE IF NOT EXISTS dm_local_files(path TEXT PRIMARY KEY,conv_id TEXT NOT NULL,generation INTEGER NOT NULL,"
"busy INTEGER NOT NULL DEFAULT 0,retry_at INTEGER NOT NULL DEFAULT 0);"
"UPDATE dm_local_files SET busy=0;";
if (sqlite3_exec(inst->topo_sqlite_db, sql, NULL, NULL, NULL) != SQLITE_OK) return history_error(inst, "init", NULL);
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM history: initialized persistent generations, replay records and file cleanup");
return 0;
}
uint64_t dm_history_generation(struct UTUN_INSTANCE* inst, const char* conv) {
sqlite3_stmt* st = NULL;
if (!inst || !inst->topo_sqlite_db || !conv || sqlite3_prepare_v2(inst->topo_sqlite_db,
"SELECT COALESCE(h.generation,1) FROM dm_conversations c LEFT JOIN dm_history h USING(conv_id) WHERE c.conv_id=?",
-1, &st, NULL) != SQLITE_OK) { history_error(inst, "generation prepare", conv); return 0; }
sqlite3_bind_text(st, 1, conv, -1, SQLITE_STATIC);
int rc = sqlite3_step(st);
uint64_t generation = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 0) : 0;
if (rc != SQLITE_ROW && rc != SQLITE_DONE) history_error(inst, "generation read", conv);
sqlite3_finalize(st);
return generation;
}
int dm_history_current(struct UTUN_INSTANCE* inst, const char* conv, uint64_t generation) {
sqlite3_stmt* st = NULL;
if (!inst || !inst->topo_sqlite_db || !conv || !generation || sqlite3_prepare_v2(inst->topo_sqlite_db,
"SELECT 1 FROM dm_conversations c LEFT JOIN dm_history h USING(conv_id)"
" WHERE c.conv_id=? AND COALESCE(h.generation,1)=? AND COALESCE(h.hidden,0)=0", -1, &st, NULL) != SQLITE_OK) {
history_error(inst, "current prepare", conv); return 0;
}
sqlite3_bind_text(st, 1, conv, -1, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)generation);
int rc = sqlite3_step(st), current = rc == SQLITE_ROW;
if (rc != SQLITE_ROW && rc != SQLITE_DONE) history_error(inst, "current read", conv);
sqlite3_finalize(st);
if (!current) DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM history: stale/hidden target conv=%s generation=%llu", conv,
(unsigned long long)generation);
return current;
}
int dm_history_show(struct UTUN_INSTANCE* inst, const char* conv) {
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(inst->topo_sqlite_db, "UPDATE dm_history SET hidden=0 WHERE conv_id=?", -1, &st, NULL) != SQLITE_OK)
return history_error(inst, "show prepare", conv);
sqlite3_bind_text(st, 1, conv, -1, SQLITE_STATIC);
int rc = sqlite3_step(st); sqlite3_finalize(st);
if (rc != SQLITE_DONE) return history_error(inst, "show", conv);
return 0;
}
int dm_history_deleted(struct UTUN_INSTANCE* inst, const uint8_t* body, size_t len, int dir, uint8_t* media_receipt) {
uint64_t conv, seq; char id[64];
if (media_receipt) memset(media_receipt, 0, DM_MEDIA_RECEIPT_SIZE);
size_t actual;
if (!dm_msg_len(body, len, &actual) || actual != len) return history_error(inst, "invalid replay body", NULL);
memcpy(&conv, body, 8); memcpy(&seq, body + 8, 8);
snprintf(id, sizeof(id), "%llu", (unsigned long long)conv);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(inst->topo_sqlite_db,
"SELECT body_hash,media_receipt FROM dm_deleted_messages WHERE conv_id=? AND dir=? AND seq=?", -1, &st, NULL) != SQLITE_OK)
return history_error(inst, "replay prepare", id);
sqlite3_bind_text(st, 1, id, -1, SQLITE_STATIC); sqlite3_bind_int(st, 2, dir); sqlite3_bind_int64(st, 3, (sqlite3_int64)seq);
int rc = sqlite3_step(st), result = rc == SQLITE_DONE ? 0 : -1;
if (rc == SQLITE_ROW) {
uint8_t hash[32]; SHA256(body, len, hash);
if (sqlite3_column_bytes(st, 0) == sizeof(hash) && !memcmp(hash, sqlite3_column_blob(st, 0), sizeof(hash))) {
result = 1;
if (media_receipt && sqlite3_column_bytes(st, 1) == DM_MEDIA_RECEIPT_SIZE)
memcpy(media_receipt, sqlite3_column_blob(st, 1), DM_MEDIA_RECEIPT_SIZE);
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM history: deleted replay conv=%s dir=%d seq=%llu", id, dir, (unsigned long long)seq);
} else DEBUG_WARN(DEBUG_CATEGORY_DM, "DM history: conflicting deleted message conv=%s dir=%d seq=%llu",
id, dir, (unsigned long long)seq);
} else if (rc != SQLITE_DONE) history_error(inst, "replay read", id);
sqlite3_finalize(st);
return result;
}
/* Каноническое тело восстанавливается из сохранённых полей; у file оно уже есть в dm_files. */
static int history_remember(struct UTUN_INSTANCE* inst, const char* conv) {
sqlite3_stmt* rows = NULL; sqlite3_stmt* put = NULL;
const char* sql = "SELECT m.dir,m.seq,m.ts,m.author,m.ct,m.data,m.sig,f.body,f.receipt"
" FROM dm_messages m LEFT JOIN dm_files f ON f.conv_id=m.conv_id AND f.dir=m.dir AND f.seq=m.seq WHERE m.conv_id=?";
if (sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &rows, NULL) != SQLITE_OK ||
sqlite3_prepare_v2(inst->topo_sqlite_db,
"INSERT INTO dm_deleted_messages(conv_id,dir,seq,body_hash,media_id,media_receipt) VALUES(?,?,?,?,?,?)",
-1, &put, NULL) != SQLITE_OK) goto fail;
sqlite3_bind_text(rows, 1, conv, -1, SQLITE_STATIC);
int rc;
while ((rc = sqlite3_step(rows)) == SQLITE_ROW) {
uint8_t body[DM_MSG_FIXED_HDR + 255 + 2 + DM_DATA_MAX + 16 + DM_MEDIA_DESC_SIZE + DM_MSG_SIG_SIZE], hash[32];
size_t length = 0;
uint64_t number = strtoull(conv, NULL, 10), seq = (uint64_t)sqlite3_column_int64(rows, 1);
uint64_t ts = (uint64_t)sqlite3_column_int64(rows, 2), author = (uint64_t)sqlite3_column_int64(rows, 3);
int ct = sqlite3_column_bytes(rows, 4), data = sqlite3_column_bytes(rows, 5), dir = sqlite3_column_int(rows, 0);
const char* type = (const char*)sqlite3_column_text(rows, 4);
if (ct == 4 && type && !memcmp(type, "file", 4)) {
int bytes = sqlite3_column_bytes(rows, 7);
if (bytes <= 0 || bytes > sizeof(body)) goto fail;
length = (size_t)bytes; memcpy(body, sqlite3_column_blob(rows, 7), length);
} else {
if (ct > 255 || data < 16 || data > DM_DATA_MAX + 16 || sqlite3_column_bytes(rows, 6) != 64) goto fail;
memcpy(body, &number, 8); memcpy(body + 8, &seq, 8); memcpy(body + 16, &ts, 8); memcpy(body + 24, &author, 8);
body[32] = (uint8_t)ct; memcpy(body + 33, type, (size_t)ct);
uint16_t size = (uint16_t)data; memcpy(body + 33 + ct, &size, 2);
memcpy(body + 35 + ct, sqlite3_column_blob(rows, 5), (size_t)data);
memcpy(body + 35 + ct + data, sqlite3_column_blob(rows, 6), 64); length = (size_t)(35 + ct + data + 64);
}
size_t actual;
if (!dm_msg_len(body, length, &actual) || actual != length) goto fail;
SHA256(body, length, hash);
sqlite3_reset(put); sqlite3_clear_bindings(put);
sqlite3_bind_text(put, 1, conv, -1, SQLITE_STATIC); sqlite3_bind_int(put, 2, dir);
sqlite3_bind_int64(put, 3, (sqlite3_int64)seq); sqlite3_bind_blob(put, 4, hash, sizeof(hash), SQLITE_TRANSIENT);
const uint8_t* media = dm_message_media(body, length);
if (media) sqlite3_bind_blob(put, 5, media, 16, SQLITE_TRANSIENT);
if (sqlite3_column_bytes(rows, 8) == DM_MEDIA_RECEIPT_SIZE)
sqlite3_bind_blob(put, 6, sqlite3_column_blob(rows, 8), DM_MEDIA_RECEIPT_SIZE, SQLITE_TRANSIENT);
if (sqlite3_step(put) != SQLITE_DONE) goto fail;
}
if (rc != SQLITE_DONE) goto fail;
sqlite3_finalize(rows); sqlite3_finalize(put);
return 0;
fail:
sqlite3_finalize(rows); sqlite3_finalize(put);
return history_error(inst, "remember deleted bodies", conv);
}
int dm_history_delete(struct UTUN_INSTANCE* inst, const char* conv, uint64_t generation) {
if (!inst || !inst->dm || !inst->dm_media || generation >= INT64_MAX || !dm_history_current(inst, conv, generation)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM history: delete rejected conv=%s generation=%llu", conv ? conv : "",
(unsigned long long)generation); return -1;
}
sqlite3* db = inst->topo_sqlite_db; sqlite3_stmt* st = NULL;
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) return history_error(inst, "begin delete", conv);
if (history_remember(inst, conv) || dm_media_forget_prepare(inst, conv, generation)) goto fail;
const char* statements[] = {
"DELETE FROM dm_outbox WHERE conv_id=?", "DELETE FROM dm_messages WHERE conv_id=?",
"INSERT INTO dm_history(conv_id,generation,hidden) VALUES(?,?,1)"
" ON CONFLICT(conv_id) DO UPDATE SET generation=excluded.generation,hidden=1"
};
for (size_t i = 0; i < sizeof(statements) / sizeof(statements[0]); i++) {
if (sqlite3_prepare_v2(db, statements[i], -1, &st, NULL) != SQLITE_OK) goto fail;
sqlite3_bind_text(st, 1, conv, -1, SQLITE_STATIC);
if (i == 2) sqlite3_bind_int64(st, 2, (sqlite3_int64)(generation + 1));
int rc = sqlite3_step(st); sqlite3_finalize(st); st = NULL;
if (rc != SQLITE_DONE) goto fail;
}
if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto fail;
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM history: deleted conv=%s generation=%llu -> %llu; counters retained", conv,
(unsigned long long)generation, (unsigned long long)(generation + 1));
dm_media_forget_committed(inst, conv);
dm_cancel_pending_history(inst, conv);
return 0;
fail:
history_error(inst, "delete rolled back", conv); sqlite3_finalize(st);
if (sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL) != SQLITE_OK) history_error(inst, "rollback delete", conv);
return -1;
}
src/dm/dm_history.h
+21
View File
@@ -0,0 +1,21 @@
/* Локальная история DM: поколения, скрытые беседы и точные отпечатки удалённых сообщений.
* Все операции выполняются в uasync. Удаление сохраняет счётчики dm_conversations. */
#ifndef DM_HISTORY_H
#define DM_HISTORY_H
#include <stdint.h>
#include <stddef.h>
#ifdef __cplusplus
extern "C" {
#endif
struct UTUN_INSTANCE;
int dm_history_init(struct UTUN_INSTANCE* inst);
uint64_t dm_history_generation(struct UTUN_INSTANCE* inst, const char* conv);
int dm_history_current(struct UTUN_INSTANCE* inst, const char* conv, uint64_t generation);
int dm_history_show(struct UTUN_INSTANCE* inst, const char* conv);
/* 1=точный удалённый повтор, 0=не удалено, -1=конфликт/ошибка. receipt может быть NULL. */
int dm_history_deleted(struct UTUN_INSTANCE* inst, const uint8_t* body, size_t len, int dir, uint8_t* media_receipt);
int dm_history_delete(struct UTUN_INSTANCE* inst, const char* conv, uint64_t generation);
#ifdef __cplusplus
}
#endif
#endif
src/dm/dm_media.c
+189 -5
View File
@@ -12,6 +12,7 @@
#include "../../lib/debug_config.h"
#include "../../lib/json_flat.h"
#include "dm_media_priv.h"
#include "dm_history.h"
#include "dm_crypto.h"
#include "dm_mailbox.h"
#include "dm_mailbox_media.h"
@@ -34,6 +35,7 @@ struct dm_media_state {
struct dm_file_job* jobs;
void* timer;
int closing;
int cleanup_active;
};
struct dm_file_job {
@@ -48,6 +50,8 @@ struct dm_file_job {
int remove_source;
char input[1024], cipher[1024], plain[1024], temp[1024], output[1024];
int error;
uint64_t generation;
unsigned held;
};
sqlite3_stmt* dm_media_prepare(struct UTUN_INSTANCE* inst, const char* sql) {
@@ -152,6 +156,129 @@ static int dm_file_remove(const char* path) {
return 0;
}
/* Манифест хранит только пути приложения. Поколение делает очередь удаления durable. */
int dm_media_file_hold(struct UTUN_INSTANCE* inst, const char* conv, uint64_t generation, const char* path) {
if (!path || !path[0] || !dm_history_current(inst, conv, generation)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM files: hold rejected conv=%s path=%s", conv ? conv : "", path ? path : ""); return -1;
}
sqlite3_stmt* st = dm_media_prepare(inst,
"INSERT INTO dm_local_files(path,conv_id,generation,busy) VALUES(?,?,?,1) ON CONFLICT(path) DO UPDATE SET busy=busy+1"
" WHERE conv_id=excluded.conv_id AND generation=excluded.generation AND busy>=0");
if (!st) return -1;
sqlite3_bind_text(st, 1, path, -1, SQLITE_STATIC); sqlite3_bind_text(st, 2, conv, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 3, (sqlite3_int64)generation);
if (dm_media_step(inst, st)) return -1;
if (sqlite3_changes(inst->topo_sqlite_db) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: path belongs to another task/generation path=%s", path); return -1;
}
DEBUG_TRACE(DEBUG_CATEGORY_DM, "DM files: hold conv=%s generation=%llu path=%s", conv, (unsigned long long)generation, path);
return 0;
}
void dm_media_file_release(struct UTUN_INSTANCE* inst, const char* path) {
sqlite3_stmt* st = dm_media_prepare(inst, "UPDATE dm_local_files SET busy=busy-1 WHERE path=? AND busy>0");
if (!st) return;
sqlite3_bind_text(st, 1, path, -1, SQLITE_STATIC);
if (!dm_media_step(inst, st) && sqlite3_changes(inst->topo_sqlite_db) != 1)
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: release without owner path=%s", path);
}
struct dm_cleanup_job { struct dm_media_state* media; char path[1024]; int error; };
static void dm_cleanup_pump(struct dm_media_state* media);
static void dm_cleanup_work(void* arg) {
struct dm_cleanup_job* job = arg;
job->error = dm_file_remove(job->path);
if (!job->error) job->error = dm_dir_sync(job->path);
}
static void dm_cleanup_done(void* arg, int error) {
struct dm_cleanup_job* job = arg;
struct dm_media_state* media = job->media;
media->cleanup_active = 0;
int success = !error && !job->error;
sqlite3_stmt* st = dm_media_prepare(media->inst, success ? "DELETE FROM dm_local_files WHERE path=? AND busy=-1" :
"UPDATE dm_local_files SET busy=0,retry_at=?2 WHERE path=?1 AND busy=-1");
if (st && !success) sqlite3_bind_int64(st, 2, (sqlite3_int64)(ntp_time_get_seconds(media->inst) + 30));
if (st) { sqlite3_bind_text(st, 1, job->path, -1, SQLITE_STATIC); if (dm_media_step(media->inst, st)) success = 0; }
else success = 0;
if (success) DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM files: removed deleted history path=%s", job->path);
else DEBUG_WARN(DEBUG_CATEGORY_DM, "DM files: cleanup retained for retry path=%s callback=%d worker=%d", job->path, error, job->error);
u_free(job);
if (success && !media->closing) dm_cleanup_pump(media);
}
static void dm_cleanup_pump(struct dm_media_state* media) {
if (media->closing || media->cleanup_active) return;
sqlite3_stmt* st = dm_media_prepare(media->inst,
"SELECT f.path FROM dm_local_files f JOIN dm_history h USING(conv_id)"
" WHERE f.generation<h.generation AND f.busy<=0 AND f.retry_at<=? ORDER BY f.retry_at,f.path LIMIT 1");
if (!st) return;
sqlite3_bind_int64(st, 1, (sqlite3_int64)ntp_time_get_seconds(media->inst));
int rc = sqlite3_step(st);
if (rc != SQLITE_ROW) {
if (rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: cleanup scan failed rc=%d", rc);
sqlite3_finalize(st); return;
}
struct dm_cleanup_job* job = u_calloc(1, sizeof(*job));
if (!job) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: cleanup allocation failed"); sqlite3_finalize(st); return; }
const char* path = (const char*)sqlite3_column_text(st, 0);
if (!path || strlen(path) >= sizeof(job->path)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: invalid cleanup path"); sqlite3_finalize(st); u_free(job); return;
}
job->media = media; snprintf(job->path, sizeof(job->path), "%s", path); sqlite3_finalize(st);
st = dm_media_prepare(media->inst, "UPDATE dm_local_files SET busy=-1 WHERE path=? AND busy<=0");
if (!st) { u_free(job); return; }
sqlite3_bind_text(st, 1, job->path, -1, SQLITE_STATIC);
if (dm_media_step(media->inst, st)) { u_free(job); return; }
media->cleanup_active = 1;
media_async_submit(media->inst->media_async, media->inst->ua, dm_cleanup_work, job, dm_cleanup_done, job);
}
int dm_media_forget_prepare(struct UTUN_INSTANCE* inst, const char* conv, uint64_t generation) {
struct dm_media_state* media = inst->dm_media;
sqlite3_stmt* rows = dm_media_prepare(inst, "SELECT id FROM dm_files WHERE conv_id=?");
if (!rows) return -1;
sqlite3_bind_text(rows, 1, conv, -1, SQLITE_STATIC);
int rc, result = 0;
while ((rc = sqlite3_step(rows)) == SQLITE_ROW) {
if (sqlite3_column_bytes(rows, 0) != 16) { result = -1; break; }
const char* suffixes[] = {".file", ".cipher", ".encrypt", ".download", ".decrypt"};
for (unsigned i = 0; i < sizeof(suffixes) / sizeof(suffixes[0]); i++) {
char path[1024];
if (dm_file_path(media, 0, sqlite3_column_blob(rows, 0), suffixes[i], path, sizeof(path))) { result = -1; break; }
sqlite3_stmt* st = dm_media_prepare(inst,
"INSERT INTO dm_local_files(path,conv_id,generation) VALUES(?,?,?) ON CONFLICT(path) DO NOTHING");
if (!st) { result = -1; break; }
sqlite3_bind_text(st, 1, path, -1, SQLITE_STATIC); sqlite3_bind_text(st, 2, conv, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 3, (sqlite3_int64)generation);
if (dm_media_step(inst, st)) { result = -1; break; }
}
if (result) break;
}
sqlite3_finalize(rows);
if (result || rc != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: delete manifest failed conv=%s rc=%d", conv, rc); return -1; }
sqlite3_stmt* st = dm_media_prepare(inst, "DELETE FROM dm_files WHERE conv_id=?");
if (!st) return -1;
sqlite3_bind_text(st, 1, conv, -1, SQLITE_STATIC);
return dm_media_step(inst, st);
}
void dm_media_forget_committed(struct UTUN_INSTANCE* inst, const char* conv) {
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT media_id FROM dm_deleted_messages WHERE conv_id=? AND media_id IS NOT NULL");
if (st) {
sqlite3_bind_text(st, 1, conv, -1, SQLITE_STATIC);
int rc;
while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
if (sqlite3_column_bytes(st, 0) == 16) file_transfer_cancel_object(inst->file_transfer, sqlite3_column_blob(st, 0));
else DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: damaged deleted media ID conv=%s", conv);
}
if (rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM files: cancellation scan failed conv=%s rc=%d", conv, rc);
sqlite3_finalize(st);
}
dm_cleanup_pump(inst->dm_media);
}
/* Подписанная роль и живые мемберы в одной области; тип NAT не даёт полномочий. */
int dm_media_scope(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t node,
const char* role, uint64_t author, uint64_t recipient) {
@@ -276,6 +403,21 @@ int dm_media_accept_receipt(struct UTUN_INSTANCE* inst, const uint8_t* r) {
if (!st) return -1;
sqlite3_bind_blob(st, 1, r + 40, 16, SQLITE_STATIC);
int rc = sqlite3_step(st);
if (rc == SQLITE_DONE) {
sqlite3_finalize(st);
st = dm_media_prepare(inst,
"SELECT body_hash FROM dm_deleted_messages WHERE conv_id=? AND dir=1 AND seq=? AND media_id=?");
if (!st) return -1;
uint64_t conv, seq; char id[64]; memcpy(&conv, r + 8, 8); memcpy(&seq, r + 16, 8);
snprintf(id, sizeof(id), "%llu", (unsigned long long)conv);
sqlite3_bind_text(st, 1, id, -1, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)seq);
sqlite3_bind_blob(st, 3, r + 40, 16, SQLITE_STATIC);
rc = sqlite3_step(st);
int exact = rc == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 32 && !memcmp(sqlite3_column_blob(st, 0), r + 56, 32);
sqlite3_finalize(st);
if (exact) { DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM files: late receipt for deleted file conv=%s", id); return 0; }
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM files: receipt for unknown/conflicting deleted file conv=%s", id); return -1;
}
if (rc != SQLITE_ROW || dm_media_verify_receipt(inst, r, sqlite3_column_blob(st, 0), (size_t)sqlite3_column_bytes(st, 0))) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "dm_media: receipt does not match outgoing artifact rc=%d", rc);
sqlite3_finalize(st);
@@ -325,10 +467,29 @@ static struct dm_file_job* dm_job_find(struct dm_media_state* media, const uint8
return NULL;
}
static int dm_job_paths(struct dm_file_job* job, const char** paths) {
if (job->work != DM_FILE_PREPARE && job->work != DM_FILE_RECEIVE) return 0;
paths[0] = job->cipher; paths[1] = job->temp;
paths[2] = job->work == DM_FILE_PREPARE ? job->plain : job->output;
if (job->work == DM_FILE_PREPARE && job->remove_source) { paths[3] = job->input; return 4; }
return 3;
}
static int dm_job_hold(struct dm_file_job* job) {
const char* paths[4]; int count = dm_job_paths(job, paths);
for (int i = 0; i < count; i++) {
if (dm_media_file_hold(job->media->inst, job->conv, job->generation, paths[i])) return -1;
job->held++;
}
return 0;
}
static void dm_job_free(struct dm_file_job* job) {
struct dm_file_job** p = &job->media->jobs;
while (*p && *p != job) p = &(*p)->next;
if (*p) *p = job->next;
const char* paths[4]; dm_job_paths(job, paths);
for (unsigned i = 0; i < job->held; i++) dm_media_file_release(job->media->inst, paths[i]);
OPENSSL_cleanse(job->key, sizeof(job->key));
u_free(job);
}
@@ -425,6 +586,10 @@ static void dm_file_done(void* arg, int error) {
}
int custody = job->work == DM_FILE_CUSTODY;
if (error || job->error || media->closing) goto fail;
if (!custody && !dm_history_current(inst, job->conv, job->generation)) {
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM files: discard deleted preparation conv=%s generation=%llu work=%d", job->conv,
(unsigned long long)job->generation, job->work); goto fail;
}
if (job->work == DM_FILE_PREPARE) {
uint8_t data[DM_MEDIA_DESC_SIZE + ATTACHMENT_WIRE_MAX];
memcpy(data, job->id, 16); memcpy(data + 16, &job->size, 8); memcpy(data + 24, job->hash, 32);
@@ -523,8 +688,11 @@ static void dm_file_pull(struct dm_media_state* media, int custody, uint64_t gro
dm_file_path(media, custody, job->id, ".decrypt", job->temp, sizeof(job->temp)) ||
dm_file_path(media, custody, job->id, custody ? ".cipher" : ".file", job->output, sizeof(job->output))) { u_free(job); return; }
if (!custody) {
snprintf(job->conv, sizeof(job->conv), "%llu", (unsigned long long)dm_derive_conv_id(media->inst->node_id, author));
job->generation = dm_history_generation(media->inst, job->conv);
if (dm_job_hold(job)) { dm_job_free(job); return; }
sqlite3_stmt* st = dm_media_prepare(media->inst, "SELECT x25519_pubkey FROM nodes WHERE node_id=?");
if (!st) { u_free(job); return; }
if (!st) { dm_job_free(job); return; }
sqlite3_bind_int64(st, 1, (sqlite3_int64)author);
int rc = sqlite3_step(st);
uint8_t pub[32];
@@ -533,7 +701,7 @@ static void dm_file_pull(struct dm_media_state* media, int custody, uint64_t gro
sqlite3_finalize(st);
if (!found || dm_derive_content_key(media->inst->my_keys.private_key, pub, job->key)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: target key unavailable author=%llu", (unsigned long long)author);
u_free(job);
dm_job_free(job);
return;
}
}
@@ -590,6 +758,8 @@ int dm_send_file(struct UTUN_INSTANCE* inst, const char* conv, const char* path,
struct dm_file_job* job = u_calloc(1, sizeof(*job));
if (!job) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: preparation allocation failed"); return -1; }
job->media = media; job->work = DM_FILE_PREPARE; job->author = inst->node_id;
job->generation = dm_history_generation(inst, conv);
if (!dm_history_current(inst, conv, job->generation)) { u_free(job); return -1; }
snprintf(job->conv, sizeof(job->conv), "%s", conv); job->info = *info; job->remove_source = remove_source;
snprintf(job->input, sizeof(job->input), "%s", path);
sqlite3_stmt* st = dm_media_prepare(inst, "SELECT peer_x25519 FROM dm_conversations WHERE conv_id=?");
@@ -606,6 +776,7 @@ int dm_send_file(struct UTUN_INSTANCE* inst, const char* conv, const char* path,
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: preparation keys/paths failed conv=%s", conv);
OPENSSL_cleanse(job->key, sizeof(job->key)); u_free(job); return -1;
}
if (dm_job_hold(job)) { dm_job_free(job); return -1; }
job->next = media->jobs; media->jobs = job;
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: preparing conv=%s kind=%d name=%s", conv, info->kind, info->name);
chat_event_attachment(inst, 1, conv, job->id, 0);
@@ -630,9 +801,12 @@ static void dm_source_cleaned(void* arg, int error) {
void dm_send_file_trampoline(void* arg) {
struct dm_file_req* req = arg;
if (!req) return;
if (dm_send_file(req->inst, req->conv, req->path, &req->info, req->remove_source)) {
uint64_t generation = req->generation ? req->generation : dm_history_generation(req->inst, req->conv);
int current = dm_history_current(req->inst, req->conv, generation);
if (!current || dm_send_file(req->inst, req->conv, req->path, &req->info, req->remove_source)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM files: send rejected conv=%s generation=%llu", req->conv, (unsigned long long)generation);
const uint8_t rejected[16] = {0};
chat_event_attachment(req->inst, 1, req->conv, rejected, 2);
if (current) chat_event_attachment(req->inst, 1, req->conv, rejected, 2);
if (req->remove_source) {
media_async_submit(req->inst ? req->inst->media_async : NULL, req->inst ? req->inst->ua : NULL,
dm_source_cleanup, req, dm_source_cleaned, req);
@@ -831,6 +1005,15 @@ static void dm_media_offer(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t
uint8_t receipt[DM_RECEIPT_SIZE];
if (dm_accept_message(inst, p, len, receipt)) return;
dm_mailbox_receipt(inst, receipt);
uint8_t file_receipt[DM_MEDIA_RECEIPT_SIZE];
int deleted = dm_history_deleted(inst, p, len, 0, file_receipt);
if (deleted) {
if (deleted > 0 && !memcmp(file_receipt, "DMFILE01", 8)) {
dm_media_send(inst, group, author, DM_MEDIA_RECEIPT, file_receipt, sizeof(file_receipt));
dm_mailbox_media_receipt(inst, file_receipt);
}
return;
}
sqlite3_stmt* st = dm_media_prepare(inst, "UPDATE dm_files SET holder=?,expires=?,group_id=?,retry_at=0 WHERE id=? AND ready<=0");
if (!st) return;
sqlite3_bind_int64(st, 1, (sqlite3_int64)holder); sqlite3_bind_int64(st, 2, (sqlite3_int64)expiry);
@@ -983,6 +1166,7 @@ static void dm_media_tick(void* arg) {
uint64_t now = ntp_time_get_seconds(media->inst);
dm_files_tick(media, now);
dm_custody_tick(media, now);
dm_cleanup_pump(media);
media->timer = uasync_set_timeout(media->inst->ua, 10000, media, dm_media_tick, "dm_media");
if (!media->timer) DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: retry timer allocation failed");
}
@@ -1043,7 +1227,7 @@ void dm_media_destroy(struct UTUN_INSTANCE* inst) {
if (!inst || !inst->dm_media) return;
dm_media_quiesce(inst);
struct dm_media_state* media = inst->dm_media;
if (media->jobs) {
if (media->jobs || media->cleanup_active) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "dm_media: destroy before media_async workers drained");
return;
}
src/dm/dm_media.h
+8
View File
@@ -37,6 +37,7 @@ int dm_send_file(struct UTUN_INSTANCE* inst, const char* conv, const char* path,
struct dm_file_req {
struct UTUN_INSTANCE* inst;
char conv[64];
uint64_t generation;
char path[1024];
struct attachment_info info;
int remove_source; /* Только принадлежащий приложению временный файл. */
@@ -50,6 +51,13 @@ uint64_t dm_media_cipher_size(uint64_t plain_size);
int dm_media_verify_receipt(struct UTUN_INSTANCE* inst, const uint8_t* receipt, const uint8_t* msg, size_t len);
int dm_media_accept_receipt(struct UTUN_INSTANCE* inst, const uint8_t* receipt);
/* Удаление: prepare внутри транзакции, committed после commit (callbacks отмены синхронны). */
int dm_media_forget_prepare(struct UTUN_INSTANCE* inst, const char* conv, uint64_t generation);
void dm_media_forget_committed(struct UTUN_INSTANCE* inst, const char* conv);
/* Владение путями worker: hold до запуска, release после завершения. Уборка ждёт busy=0. */
int dm_media_file_hold(struct UTUN_INSTANCE* inst, const char* conv, uint64_t generation, const char* path);
void dm_media_file_release(struct UTUN_INSTANCE* inst, const char* path);
/* Общие операции control для endpoint/M и mailbox/S. Все пакеты подписаны. */
int dm_media_send(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer,
uint8_t cmd, const uint8_t* body, size_t len);
src/media_async/attachment_send.c
+29 -3
View File
@@ -11,6 +11,7 @@
#include "../chat/chat_core.h"
#include "../chat/chat_event.h"
#include "../dm/dm_media.h"
#include "../dm/dm_history.h"
#include "../video/video.h"
#include "media_async.h"
#include "voice_file.h"
@@ -57,18 +58,29 @@ done:
}
/* Один выбор транспорта после общей подготовки; групповой индекс и PM crypto не смешиваются. */
static void attachment_free(struct attachment_send_req* req) {
if (req->dm_paths_held & 1) dm_media_file_release(req->inst, req->output);
if (req->dm_paths_held & 2) dm_media_file_release(req->inst, req->source);
u_free(req);
}
static void attachment_publish(struct attachment_send_req* req, int error) {
attachment_release_pcm(req); /* Submit мог завершиться ошибкой до запуска work. */
int abandoned = req->is_dm && req->dm_generation && !dm_history_current(req->inst, req->target, req->dm_generation);
if (abandoned) {
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "attachment: discarded deleted PM preparation target=%s generation=%llu",
req->target, (unsigned long long)req->dm_generation); error = MEDIA_ASYNC_CANCELLED;
}
if (req->temporary_source && remove(req->source) && errno != ENOENT) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment: abandoned source cleanup failed path=%s", req->source);
error = -1;
}
if (error || req->error || attachment_validate(&req->info)) {
chat_event_attachment(req->inst, req->is_dm, req->target, req->id, 2);
if (!abandoned || req->dm_paths_held) chat_event_attachment(req->inst, req->is_dm, req->target, req->id, 2);
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment: preparation failed target=%s error=%d worker=%d", req->target, error, req->error);
chat_event_submitted(req->inst, req->request_id, 0);
if (remove(req->output) && errno != ENOENT) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment: output cleanup failed path=%s", req->output);
u_free(req); return;
attachment_free(req); return;
}
chat_event_attachment(req->inst, req->is_dm, req->target, req->id, 1);
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "attachment: prepared target=%s pm=%d kind=%d duration=%u dimensions=%ux%u",
@@ -77,6 +89,7 @@ static void attachment_publish(struct attachment_send_req* req, int error) {
struct dm_file_req* file = u_calloc(1, sizeof(*file));
if (file) {
file->inst = req->inst; file->info = req->info; file->remove_source = 1;
file->generation = req->dm_generation;
snprintf(file->conv, sizeof(file->conv), "%s", req->target);
snprintf(file->path, sizeof(file->path), "%s", req->output);
dm_send_file_trampoline(file);
@@ -108,7 +121,7 @@ static void attachment_publish(struct attachment_send_req* req, int error) {
if (remove(req->output)) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment: rejected output cleanup failed");
}
}
u_free(req);
attachment_free(req);
}
static void attachment_video_done(void* arg, int error, const struct video_info* info) {
@@ -123,6 +136,7 @@ static void attachment_video_done(void* arg, int error, const struct video_info*
static void attachment_prepare_done(void* arg, int error) {
struct attachment_send_req* req = arg;
if (req->is_dm && !dm_history_current(req->inst, req->target, req->dm_generation)) error = MEDIA_ASYNC_CANCELLED;
if (!error && !req->error && req->transcode) {
if (!video_transcode_start(req->inst->media_async, req->inst->ua, req->source, req->output, attachment_video_done, req)) return;
error = -1;
@@ -139,6 +153,10 @@ void attachment_send_trampoline(void* arg) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment: invalid target/stopped service");
attachment_publish(req, -1); return;
}
if (req->is_dm) {
if (!req->dm_generation) req->dm_generation = dm_history_generation(req->inst, req->target);
if (!dm_history_current(req->inst, req->target, req->dm_generation)) { attachment_publish(req, -1); return; }
}
uint8_t uuid[16]; char hex[33];
if (RAND_bytes(uuid, sizeof(uuid)) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment: UUID generation failed"); attachment_publish(req, -1); return;
@@ -153,6 +171,14 @@ void attachment_send_trampoline(void* arg) {
}
if (!req->info.name[0]) snprintf(req->info.name, sizeof(req->info.name), "%s.%s", hex,
req->info.kind == ATTACHMENT_VOICE ? "opus" : "mp4");
if (req->is_dm) {
if (dm_media_file_hold(req->inst, req->target, req->dm_generation, req->output)) { attachment_publish(req, -1); return; }
req->dm_paths_held |= 1;
if (req->temporary_source) {
if (dm_media_file_hold(req->inst, req->target, req->dm_generation, req->source)) { attachment_publish(req, -1); return; }
req->dm_paths_held |= 2;
}
}
chat_event_attachment(req->inst, req->is_dm, req->target, req->id, 0);
media_async_submit(req->inst->media_async, req->inst->ua, attachment_prepare_work, req, attachment_prepare_done, req);
}
src/media_async/attachment_send.h
+2
View File
@@ -17,6 +17,8 @@ struct attachment_send_req {
struct UTUN_INSTANCE* inst;
char target[64]; /* Десятичный channel_id либо conv_id без префикса dm:, зафиксированный до подготовки. */
int is_dm;
uint64_t dm_generation; /* Захватывается до подготовки и сохраняется во всех её стадиях. */
unsigned dm_paths_held;
uint8_t reply_to[32]; /* Оригинал канального ответа, зафиксированный до подготовки. */
uint64_t request_id; /* CHAT_EVT_MSG_SUBMITTED после публикации либо ошибки. */
struct attachment_info info;