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

Scope media registrations and active roles to their CHAT group

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 19:07:51 +01:00
1 parent ed9e171683
commit 6be77de7f6
3 files changed
+38 -18
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/media_delivery/media_delivery.c
+33 -17
View File
@@ -580,30 +580,30 @@ int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id,
/* ── subcommand handlers ── */
static void md_handle_serve_reg(struct media_delivery_ctx* md, uint64_t from_node,
static void md_handle_serve_reg(struct media_delivery_ctx* md, uint64_t group_id, uint64_t from_node,
const uint8_t* data, size_t len) {
if (len < sizeof(struct media_pkt_serve_reg)) return;
struct media_pkt_serve_reg* pkt = (struct media_pkt_serve_reg*)data;
const struct media_pkt_serve_reg* pkt = (const void*)data;
if (len != sizeof(*pkt) || pkt->group_id != group_id) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: SERVE_REG invalid size/group bytes=%zu group=%llx", MD_ID, len,
(unsigned long long)group_id); return;
}
if (!md->served_nodes) {
md->served_nodes = queue_new(md->inst->ua, 64, offsetof(struct media_served_node, node_id), 8, "md_served");
md->served_nodes = queue_new(md->inst->ua, 64, offsetof(struct media_served_node, node_id), 16, "md_served");
if (!md->served_nodes) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_new(served_nodes) failed", MD_ID); return; }
}
struct ll_entry* e = queue_find_data_by_index(md->served_nodes, &from_node);
uint64_t key[2] = {from_node, group_id};
struct ll_entry* e = queue_find_data_by_index(md->served_nodes, key);
if (!e) {
struct ll_entry* qe = queue_entry_new(sizeof(struct media_served_node));
if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_entry_new(served_node) failed", MD_ID); return; }
struct media_served_node* sn = (struct media_served_node*)qe->data;
memset(sn, 0, sizeof(*sn));
sn->node_id = from_node; sn->group_id = pkt->group_id;
sn->node_id = from_node; sn->group_id = group_id;
sn->joined_at = (int64_t)time(NULL);
memcpy(qe->data, &from_node, 8);
queue_data_put_with_index(md->served_nodes, qe);
}
uint8_t ack[9] = { MEDIA_SUBCMD_SERVE_ACK, 0, 0 };
memcpy(ack + 1, &pkt->group_id, 8);
(void)from_node;
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: SERVE_REG node=0x%016llx group=%016llx",
MD_ID, (unsigned long long)from_node, (unsigned long long)pkt->group_id);
}
@@ -1121,7 +1121,8 @@ static int md_group_role(struct media_delivery_ctx* md, uint64_t group_id, uint6
struct TOPO_GROUP* g = md->inst->topo_groups ? topo_groups_find(md->inst->topo_groups, group_id) : NULL;
if (!g || g->group_type != TOPO_GROUP_TYPE_CHAT || g->stopping || !g->channel_id[0]) return 0;
char sql[256];
sqlite3_snprintf(sizeof(sql), sql, "SELECT adm_tags FROM \"peers_%w\" WHERE node_id=? AND admitted=1", g->channel_id);
sqlite3_snprintf(sizeof(sql), sql,
"SELECT adm_tags FROM \"peers_%w\" WHERE node_id=? AND source=0 AND deleted=0 AND admitted=1", g->channel_id);
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(md->db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id);
@@ -1212,7 +1213,8 @@ void media_delivery_refresh_roles(struct UTUN_INSTANCE* inst) {
any = 1;
md_seed_local(md, group->group_id);
char sql[256];
sqlite3_snprintf(sizeof(sql), sql, "SELECT node_id,adm_tags FROM \"peers_%w\" WHERE node_id!=? AND admitted=1", group->channel_id);
sqlite3_snprintf(sizeof(sql), sql,
"SELECT node_id,adm_tags FROM \"peers_%w\" WHERE node_id!=? AND source=0 AND deleted=0 AND admitted=1", group->channel_id);
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(md->db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)md->self_node_id);
@@ -1258,9 +1260,12 @@ static void md_on_conn_status(struct ETCP_CONN* conn, int status, void* arg) {
switch (status) {
case ETCP_CONN_STATUS_DOWN:
case ETCP_CONN_STATUS_DELETE:
if (md->served_nodes) {
struct ll_entry* e = queue_find_data_by_index(md->served_nodes, &node_id);
if (e) { queue_remove_data(md->served_nodes, e); queue_entry_free(e); }
for (struct ll_entry* e = md->served_nodes ? md->served_nodes->head : NULL; e;) {
struct ll_entry* next = e->next;
if (((struct media_served_node*)e->data)->node_id == node_id) {
queue_remove_data(md->served_nodes, e); queue_entry_free(e);
}
e = next;
}
break;
default: break;
@@ -1288,6 +1293,11 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
memcpy(&from_node, data + ROUTER_SVC_SRC_OFF, 8);
memcpy(&group_id, data + ROUTER_SVC_GROUP_OFF, 8);
if (len == ROUTER_SVC_PAYLOAD_OFF) {
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: channel closed group=%llx peer=%llx", MD_ID,
(unsigned long long)group_id, (unsigned long long)from_node);
uint64_t key[2] = {from_node, group_id};
struct ll_entry* served = md->served_nodes ? queue_find_data_by_index(md->served_nodes, key) : NULL;
if (served) { queue_remove_data(md->served_nodes, served); queue_entry_free(served); }
for (struct stream_ctx* sc = md->streams; sc;) {
struct stream_ctx* next = sc->next;
if (sc->group_id == group_id && sc->dst_node_id == from_node) stream_free(sc);
@@ -1312,13 +1322,19 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
}
switch (subcmd) {
case MEDIA_SUBCMD_SERVE_REG:
md_handle_serve_reg(md, from_node, data, len);
md_handle_serve_reg(md, group_id, from_node, data, len);
break;
case MEDIA_SUBCMD_SERVE_LEAVE:
if (len != sizeof(struct media_pkt_serve_leave)
|| ((const struct media_pkt_serve_leave*)data)->group_id != group_id) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: SERVE_LEAVE invalid size/group bytes=%zu group=%llx", MD_ID, len,
(unsigned long long)group_id); break;
}
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: SERVE_LEAVE from 0x%016llx",
MD_ID, (unsigned long long)from_node);
if (md->served_nodes) {
struct ll_entry* e = queue_find_data_by_index(md->served_nodes, &from_node);
uint64_t key[2] = {from_node, group_id};
struct ll_entry* e = queue_find_data_by_index(md->served_nodes, key);
if (e) { queue_remove_data(md->served_nodes, e); queue_entry_free(e); }
}
break;
src/media_delivery/media_delivery.h
+1 -1
View File
@@ -114,7 +114,7 @@ struct media_delivery_ctx {
int query_timeout_tb; // Дедлайн одной попытки QUERY, независимо от входящих чанков; 0 — 2с.
int dl_max_attempts; // попыток на блок (по умолч. MEDIA_DL_MAX_ATTEMPTS)
sqlite3* db; // = inst->topo_sqlite_db (кешируем для удобства)
struct ll_queue* served_nodes; // media_served_node — только суперузел
struct ll_queue* served_nodes; // media_served_node, индекс (node_id, group_id); регистрации независимы по группам
struct ll_queue* super_peers; // media_super_peer — только суперузел
struct ll_queue* downloads; // media_download — активные загрузки
struct ll_queue* block_sources; // media_block_source, заимствованный индекс по block_id.
src/media_delivery/media_download.c
+4
View File
@@ -779,6 +779,10 @@ int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id,
}
struct media_delivery_ctx* md = &inst->md;
if (!md->initialized) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: start before media_delivery init", MDL_ID); return -1; }
if (!topo_group_chat_allowed(inst, group_id, inst->node_id)) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: start denied outside active CHAT membership group=%llx", MDL_ID,
(unsigned long long)group_id); return -1;
}
char location[MEDIA_PATH_MAX], normalized[MEDIA_PATH_MAX];
uint8_t author_pubkey[32];