From f91894aa1b9bb35d90ee1c7caa4274dea0080918 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 12 Oct 2026 21:57:20 +0200 Subject: [PATCH] Expose verified stream sources and channel watch events --- src/chat/chat_event.h | 1 + src/stream/stream.c | 118 +++++++++++++++++++++++++++++++++++++++++- src/stream/stream.h | 6 +++ 3 files changed, 124 insertions(+), 1 deletion(-) diff --git a/src/chat/chat_event.h b/src/chat/chat_event.h index 468500ca..1d3c9cf6 100644 --- a/src/chat/chat_event.h +++ b/src/chat/chat_event.h @@ -79,6 +79,7 @@ extern "C" { #define CHAT_EVT_RADIO_MUTE_RESULT 55 /* [group_id:8LE][node_id:8LE][success:1] — локальный переключатель рации */ #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] — доступные публикации выбранного канала */ typedef void (*chat_event_handler_fn)(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len); diff --git a/src/stream/stream.c b/src/stream/stream.c index 2565c0b2..d6cd9729 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -1,12 +1,16 @@ #include #include +#include #include "../../lib/debug_config.h" #include "../../lib/mem.h" #include "../../lib/platform_compat.h" #include "../../lib/u_async.h" +#include "../../lib/strbuf.h" +#include "../../lib/sqlite3.h" #include "../utun_instance.h" #include "../chat/chat_core.h" +#include "../chat/chat_event.h" #include "stream_member.h" #include "stream_source.h" #include "stream_priv.h" @@ -50,12 +54,112 @@ struct stream_service { stream_event_fn callback; void *arg; void *reconcile; + void *sources_pending; + uint64_t sources_group; + char *sources_snapshot; + int sources_valid; int stopping, dispatching, registered, initializing; }; static void session_deliver(void *arg); static void service_reconcile(void *arg); +/* Общий каталог не выделяет media-ресурсов: профиль проверяется тем же кодом, что и subscribe. */ +int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **out) { + if (!inst || !out || *out || !group_id) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list invalid arguments"); return STREAM_ERR_ARGUMENT; + } + struct TOPO_GROUP *group = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL; + if (!group || group->stopping || group->group_type != TOPO_GROUP_TYPE_CHAT) return STREAM_WAIT; + for (const char *p = group->channel_id; *p; p++) if (*p < '0' || *p > '9') { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source namespace invalid group=%llx", (unsigned long long)group_id); + return STREAM_ERR_STATE; + } + sqlite3 *db = chat_core_get_db(inst); + if (!db) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list missing DB"); return STREAM_ERR_STATE; } + char query[320]; + snprintf(query, sizeof(query), "SELECT p.node_id,COALESCE(NULLIF(p.local_nick,''),n.name,'') " + "FROM \"peers_%s\" p LEFT JOIN nodes n ON p.node_id=n.node_id ORDER BY p.node_id", group->channel_id); + struct strbuf json = strbuf_new(); + sqlite3_stmt *statement = NULL; + int result = STREAM_OK, count = 0; + if (strbuf_addf(&json, "[") < 0) result = STREAM_ERR_MEMORY; + if (result == STREAM_OK && sqlite3_prepare_v2(db, query, -1, &statement, NULL) != SQLITE_OK) result = STREAM_ERR_STATE; + int step = SQLITE_DONE; + while (result == STREAM_OK && (step = sqlite3_step(statement)) == SQLITE_ROW) { + uint64_t node = (uint64_t)sqlite3_column_int64(statement, 0); + if (!stream_member_online(inst, group_id, node)) continue; + struct stream_member_state state; + int loaded = stream_member_load(inst, group_id, node, &state, NULL); + if (loaded == STREAM_WAIT || loaded == STREAM_ERR_NOT_MEMBER) continue; + if (loaded < 0) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "source excluded group=%llx node=%llx profile_result=%d", + (unsigned long long)group_id, (unsigned long long)node, loaded); continue; + } + if (!state.published) continue; + if (strbuf_addf(&json, "%s{\"id\":\"%llu\",\"epoch\":\"%016llx\",\"self\":%d,\"name\":", count ? "," : "", + (unsigned long long)node, (unsigned long long)state.source.source_epoch, node == inst->node_id) < 0 + || strbuf_add_json_string(&json, (const char *)sqlite3_column_text(statement, 1)) < 0 + || strbuf_addf(&json, "}") < 0) result = STREAM_ERR_MEMORY; + count++; + } + if (result == STREAM_OK && step != SQLITE_DONE) result = STREAM_ERR_STATE; + sqlite3_finalize(statement); + if (result == STREAM_OK && strbuf_addf(&json, "]") < 0) result = STREAM_ERR_MEMORY; + if (result == STREAM_OK) *out = strbuf_detach(&json); + strbuf_free(&json); + if (result != STREAM_OK) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list failed group=%llx result=%d DB=%s", + (unsigned long long)group_id, result, sqlite3_errmsg(db)); + else DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source list group=%llx count=%d", (unsigned long long)group_id, count); + return result; +} + +/* Один отложенный снимок объединяет Merkle/BGP-события; неизменный список повторно не отправляется. */ +static void sources_deliver(void *arg) { + struct stream_service *service = arg; + service->sources_pending = NULL; + if (service->stopping || !service->sources_group) return; + char *json = NULL; + int result = stream_sources_json(service->inst, service->sources_group, &json); + int valid = result == STREAM_OK || result == STREAM_WAIT; + const char *list = json ? json : "[]"; + if (service->sources_snapshot && valid == service->sources_valid && !strcmp(service->sources_snapshot, list)) { + u_free(json); return; + } + size_t length = strlen(list); + uint8_t *event = length <= INT_MAX - 9 ? u_malloc((uint32_t)(9 + length)) : NULL; + if (!event) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source event allocation failed bytes=%zu", length); u_free(json); return; + } + stream_write_be(event, service->sources_group, 8); event[8] = (uint8_t)valid; + memcpy(event + 9, list, length); + char *snapshot = json ? json : u_strdup(list); + if (!snapshot) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source snapshot allocation failed"); + u_free(service->sources_snapshot); service->sources_snapshot = snapshot; service->sources_valid = valid; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream sources changed group=%llx valid=%d bytes=%zu", + (unsigned long long)service->sources_group, valid, length); + chat_event_post(service->inst, CHAT_EVT_STREAM_SOURCES, event, (int)(length + 9)); + u_free(event); +} + +/* Наблюдение принадлежит сервису и снимается до освобождения ядра. */ +int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id) { + struct stream_service *service = inst ? inst->stream : NULL; + if (!service || service->stopping) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream source watch without service group=%llx", (unsigned long long)group_id); + return STREAM_ERR_STATE; + } + service->sources_group = group_id; + u_free(service->sources_snapshot); service->sources_snapshot = NULL; + if (service->sources_pending) uasync_call_soon_cancel(inst->ua, service->sources_pending); + service->sources_pending = group_id ? uasync_call_soon(inst->ua, service, sources_deliver) : NULL; + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source watch group=%llx", (unsigned long long)group_id); + if (group_id && !service->sources_pending) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source watch scheduling failed"); return STREAM_ERR_MEMORY; + } + return STREAM_OK; +} + /* Observer выполняет только учёт; пользовательские callbacks и отправка остаются отложенными. */ static void cache_event(void *arg, const uint8_t *wire, size_t length, int added) { struct stream_session *s = arg; @@ -268,7 +372,13 @@ static void service_reconcile(void *arg) { static void member_changed(uint64_t node_id, const char *tags, const char *channel, void *arg) { (void)tags; struct stream_service *service = arg; - if (service->stopping || !service->sessions) return; + if (service->stopping) return; + struct TOPO_GROUP *watched = service->sources_group ? stream_group(service->inst, service->sources_group) : NULL; + if (watched && channel && !strcmp(watched->channel_id, channel) && !service->sources_pending) { + service->sources_pending = uasync_call_soon(service->inst->ua, service, sources_deliver); + if (!service->sources_pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source refresh scheduling failed"); + } + if (!service->sessions) return; stream_network_membership_changed(service->network, node_id, channel); for (struct stream_session *s = service->sessions; s; s = s->next) { int own = node_id == service->inst->node_id; @@ -344,6 +454,8 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) { stream_network_destroy(service->network); service->network = NULL; if (service->registered) member_sync_remove_props_cbk(inst, member_changed, service); if (service->reconcile) uasync_call_soon_cancel(inst->ua, service->reconcile); + if (service->sources_pending) uasync_call_soon_cancel(inst->ua, service->sources_pending); + u_free(service->sources_snapshot); while (service->sessions) { struct stream_session *s = service->sessions; if (!s->closing) session_close(s, STREAM_CLOSE_SERVICE, 1, 0); @@ -360,6 +472,10 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) { void stream_cancel_group(struct UTUN_INSTANCE *inst, uint64_t group_id) { struct stream_service *service = inst ? inst->stream : NULL; if (!service) return; + if (service->sources_group == group_id && !service->sources_pending) { + service->sources_pending = uasync_call_soon(inst->ua, service, sources_deliver); + if (!service->sources_pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream cancelled group source refresh failed"); + } struct stream_session *s = service->sessions; while (s) { struct stream_session *next = s->next; diff --git a/src/stream/stream.h b/src/stream/stream.h index 6f531ab5..3f6da37b 100644 --- a/src/stream/stream.h +++ b/src/stream/stream.h @@ -60,6 +60,12 @@ int stream_service_init(struct UTUN_INSTANCE *inst); void stream_service_destroy(struct UTUN_INSTANCE *inst); void stream_cancel_group(struct UTUN_INSTANCE *inst, uint64_t group_id); void stream_set_event_callback(struct UTUN_INSTANCE *inst, stream_event_fn callback, void *arg); +/* Проверенные online-публикации: JSON с id(decimal string), epoch(hex string), name, self. + * Только uasync; *out заранее NULL, результат освобождается u_free(). WAIT — группа недоступна. */ +int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **out); +/* Один выбранный канал frontend: начальный снимок и изменения через CHAT_EVT_STREAM_SOURCES. + * group_id=0 снимает наблюдение. Не создаёт подписку/буфер и не удерживает ACTIVE. Только uasync. */ +int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id); /* Локальный owner удерживает эффективный ACTIVE; желаемый frontend-режим ядро восстанавливает после последнего close. */ int stream_service_needs_activity(const struct UTUN_INSTANCE *inst);