jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
a444f2813421262fdd4c6577aaf3d76d1de34f21
utun2/src/chat/group_avatar.c
T

446 lines
27 KiB
C
Raw Blame History

#include "group_avatar.h"
#include "chat_core_priv.h"
#include "chat_event.h"
#include "../utun_instance.h"
#include "../ntp_time.h"
#include "../member_files/member_files.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "../routing_layer/topo_group.h"
#include "../routing_layer/etcp_router.h"
#include "../transport_layer/etcp_api.h"
#include "../transport_layer/etcp.h"
#include "../../lib/mem.h"
#include "../../lib/u_async.h"
#include <openssl/sha.h>
#define GA_STATE 1
#define GA_GET 2
#define GA_DATA 3
#define GA_WIRE_VERSION 1
#define GA_STATE_BYTES 44
#define GA_DATA_HEADER 112
#define GA_STREAMS 16
#define GA_RETRY_TB 150000ULL
#define GA_EXCHANGE_TB 300000ULL
struct ga_peer {
struct ga_peer* next;
struct group_avatar* owner;
uint64_t group, peer, epoch, advertised_at, requested_at, rx_at, tx_at;
uint64_t remote_version;
uint8_t remote_hash[32];
int state_pending, get_pending, reply_pending, seen;
uint16_t tx_offset, rx_offset;
struct group_avatar_record *tx, *rx;
struct queue_waiter_handle waiter;
void* pump;
};
struct group_avatar { struct UTUN_INSTANCE* inst; struct ga_peer* peers; void* timer; unsigned tx_count, rx_count; int closing; };
static void ga_pump(void* arg);
static void ga_schedule(struct ga_peer* peer);
static struct group_avatar* ga_context(struct UTUN_INSTANCE* inst) { return inst && CC(inst) ? CC(inst)->group_avatar : NULL; }
static void ga_channel(uint64_t group, char text[32]) { snprintf(text, 32, "%llu", (unsigned long long)group); }
static int ga_message(uint64_t group, const struct group_avatar_record* record, uint8_t* bytes) {
const char domain[] = "group-avatar-v1"; size_t off = sizeof(domain) - 1;
memcpy(bytes, domain, off); mf_put64(bytes + off, group); off += 8;
mf_put64(bytes + off, record->version); off += 8;
bytes[off++] = (uint8_t)record->size; bytes[off++] = (uint8_t)(record->size >> 8);
memcpy(bytes + off, record->data, record->size); return (int)(off + record->size);
}
static int ga_verify(struct UTUN_INSTANCE* inst, uint64_t group, const struct group_avatar_record* r) {
if (!inst || !CC(inst) || !group || !r->version || r->version > INT64_MAX || r->size > GROUP_AVATAR_MAX) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT, "group avatar: invalid record group=%llu", (unsigned long long)group); return -1;
}
uint8_t hash[32], pub[32], bytes[GROUP_AVATAR_MAX + 64]; char ch[32]; ga_channel(group, ch);
SHA256(r->data, r->size, hash);
if (memcmp(hash, r->hash, 32)) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT, "group avatar: hash mismatch group=%llu version=%llu bytes=%u",
(unsigned long long)group, (unsigned long long)r->version, r->size); return -1;
}
if (topo_node_sqlite_channel_get(CC(inst)->db, ch, NULL, 0, NULL, NULL, pub, NULL) < 0) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT, "group avatar: unknown group key group=%llu", (unsigned long long)group); return -1;
}
int length = ga_message(group, r, bytes);
if (sc_ed25519_verify(pub, bytes, length, r->signature) != SC_OK) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT, "group avatar: invalid group signature group=%llu version=%llu",
(unsigned long long)group, (unsigned long long)r->version); return -1;
}
return 0;
}
int group_avatar_get(struct UTUN_INSTANCE* inst, uint64_t group, struct group_avatar_record* r) {
if (!inst || !CC(inst) || !CC(inst)->db || !r) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: invalid read"); return -1; }
memset(r, 0, sizeof(*r)); char ch[32]; ga_channel(group, ch); sqlite3_stmt* st = NULL;
const char* sql = "SELECT avatar_version,avatar_data,avatar_hash,avatar_signature FROM channels WHERE channel_id=?";
int rc = -1;
if (sqlite3_prepare_v2(CC(inst)->db, sql, -1, &st, NULL) == SQLITE_OK) {
sqlite3_bind_text(st, 1, ch, -1, SQLITE_STATIC);
if (sqlite3_step(st) == SQLITE_ROW) {
r->version = (uint64_t)sqlite3_column_int64(st, 0); int size = sqlite3_column_bytes(st, 1);
if ((!r->version && !size && !sqlite3_column_bytes(st, 2) && !sqlite3_column_bytes(st, 3)) ||
(r->version && r->version <= INT64_MAX && size <= GROUP_AVATAR_MAX &&
sqlite3_column_bytes(st, 2) == 32 && sqlite3_column_bytes(st, 3) == 64)) {
r->size = (uint16_t)size;
if (r->version) {
if (size) memcpy(r->data, sqlite3_column_blob(st, 1), size);
memcpy(r->hash, sqlite3_column_blob(st, 2), 32); memcpy(r->signature, sqlite3_column_blob(st, 3), 64);
}
rc = 0;
}
}
sqlite3_finalize(st);
}
if (!rc && r->version) rc = ga_verify(inst, group, r);
if (rc) DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: stored record unavailable/invalid group=%llu: %s",
(unsigned long long)group, sqlite3_errmsg(CC(inst)->db));
return rc;
}
static int ga_compare(uint64_t a, const uint8_t ah[32], uint64_t b, const uint8_t bh[32]) {
return a != b ? (a > b ? 1 : -1) : memcmp(ah, bh, 32);
}
static void ga_changed(struct UTUN_INSTANCE* inst, uint64_t group) {
struct group_avatar* ctx = ga_context(inst);
for (struct ga_peer* p = ctx ? ctx->peers : NULL; p; p = p->next) if (p->group == group) {
p->state_pending = 1; ga_schedule(p);
}
char ch[32]; ga_channel(group, ch); uint8_t event[33]; event[0] = (uint8_t)strlen(ch);
memcpy(event + 1, ch, event[0]); chat_event_post(inst, CHAT_EVT_CHANNEL_UPDATED, event, event[0] + 1);
}
int group_avatar_apply(struct UTUN_INSTANCE* inst, uint64_t group, const struct group_avatar_record* r) {
if (!r) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: missing candidate record"); return -1; }
if (ga_verify(inst, group, r) < 0) return -1;
struct group_avatar_record current;
if (group_avatar_get(inst, group, &current) < 0) return -1;
int newer = ga_compare(r->version, r->hash, current.version, current.hash);
if (r->version == current.version && memcmp(r->hash, current.hash, 32))
DEBUG_WARN(DEBUG_CATEGORY_CHAT, "group avatar: concurrent versions group=%llu version=%llu accept=%d",
(unsigned long long)group, (unsigned long long)r->version, newer > 0);
if (newer <= 0) return 0;
sqlite3_stmt* st = NULL; char ch[32]; ga_channel(group, ch); int rc = -1;
const char* sql = "UPDATE channels SET avatar_version=?,avatar_data=?,avatar_hash=?,avatar_signature=? WHERE channel_id=?";
if (sqlite3_prepare_v2(CC(inst)->db, sql, -1, &st, NULL) == SQLITE_OK) {
sqlite3_bind_int64(st, 1, (sqlite3_int64)r->version); sqlite3_bind_blob(st, 2, r->data, r->size, SQLITE_STATIC);
sqlite3_bind_blob(st, 3, r->hash, 32, SQLITE_STATIC); sqlite3_bind_blob(st, 4, r->signature, 64, SQLITE_STATIC);
sqlite3_bind_text(st, 5, ch, -1, SQLITE_STATIC);
if (sqlite3_step(st) == SQLITE_DONE && sqlite3_changes(CC(inst)->db) == 1) rc = 1;
sqlite3_finalize(st);
}
if (rc < 0) DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: database update failed group=%llu: %s",
(unsigned long long)group, sqlite3_errmsg(CC(inst)->db));
else {
DEBUG_INFO(DEBUG_CATEGORY_CHAT, "group avatar: saved group=%llu version=%llu bytes=%u",
(unsigned long long)group, (unsigned long long)r->version, r->size); ga_changed(inst, group);
}
return rc;
}
int group_avatar_set(struct UTUN_INSTANCE* inst, uint64_t group, const uint8_t* data, size_t size) {
if (!inst || !CC(inst) || !CC(inst)->initialized || !CC(inst)->db || size > GROUP_AVATAR_MAX ||
(size && !data) || !member_files_member(inst, group, inst->node_id, 0)) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT, "group avatar: publication denied group=%llu bytes=%zu", (unsigned long long)group, size); return -1;
}
uint8_t key[32], bytes[GROUP_AVATAR_MAX + 64]; char ch[32]; ga_channel(group, ch);
if (topo_node_sqlite_channel_get_priv(CC(inst)->db, ch, key) < 0) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT, "group avatar: publication requires group key group=%llu", (unsigned long long)group); return -1;
}
struct group_avatar_record r;
if (group_avatar_get(inst, group, &r) < 0 || r.version == INT64_MAX) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: cannot allocate version group=%llu", (unsigned long long)group); return -1;
}
uint8_t hash[32]; SHA256(size ? data : r.data, size, hash);
if (r.version && r.size == size && !memcmp(r.hash, hash, 32)) return 0;
int64_t us = ntp_time_get_us(inst); uint64_t now = us > 0 ? (uint64_t)us / 1000 : 0;
r.version = now > r.version ? now : r.version + 1; r.size = (uint16_t)size;
memcpy(r.hash, hash, 32); if (size) memcpy(r.data, data, size);
int n = ga_message(group, &r, bytes); int signed_ok = sc_ed25519_sign(key, bytes, n, r.signature); memset(key, 0, sizeof(key));
if (signed_ok != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: signing failed group=%llu", (unsigned long long)group); return -1; }
return group_avatar_apply(inst, group, &r) > 0 ? 0 : -1;
}
void group_avatar_gui_set(void* arg) {
struct group_avatar_gui_request* q = arg;
int rc = group_avatar_set(q->inst, q->group, q->data, q->size);
uint8_t result[17]; mf_put64(result, q->id); mf_put64(result + 8, q->group); result[16] = rc ? 1 : 0;
chat_event_post(q->inst, CHAT_EVT_GROUP_AVATAR_RESULT, result, sizeof(result)); u_free(q);
}
static struct ga_peer* ga_find(struct group_avatar* ctx, uint64_t group, uint64_t peer, int create) {
for (struct ga_peer* p = ctx->peers; p; p = p->next) if (p->group == group && p->peer == peer) return p;
if (!create) return NULL;
struct ga_peer* p = u_calloc(1, sizeof(*p));
if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: peer allocation failed"); return NULL; }
p->owner = ctx; p->group = group; p->peer = peer; p->state_pending = 1;
struct TOPO_GROUP* g = topo_groups_find(ctx->inst->topo_groups, group);
for (struct ll_entry* e = g && g->senders_list ? g->senders_list->head : NULL; e; e = e->next) {
struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data;
if (item && item->node_id == peer) { p->epoch = item->local_epoch; break; }
}
p->next = ctx->peers; ctx->peers = p; return p;
}
static int ga_allowed(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer) {
if (!member_files_member(inst, group, inst->node_id, 0) || !member_files_member(inst, group, peer, 0)) return 0;
struct TOPO_GROUP* g = topo_groups_find(inst->topo_groups, group);
for (struct ll_entry* e = g && g->senders_list ? g->senders_list->head : NULL; e; e = e->next) {
struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data;
if (item && item->node_id == peer && item->conn && item->ready_notified && !item->transport_down) return 1;
}
return 0;
}
static void ga_release(struct ga_peer* p, int sending) {
struct group_avatar_record** record = sending ? &p->tx : &p->rx;
if (!*record) return;
u_free(*record); *record = NULL;
if (sending) p->owner->tx_count--; else p->owner->rx_count--;
if (!p->owner->closing) for (struct ga_peer* other = p->owner->peers; other; other = other->next) {
ga_schedule(other);
if (!sending) etcp_router_resume_receive(p->owner->inst, other->group, other->peer, ETCP_RT_ID_GROUP_AVATAR);
}
}
static void ga_free(struct ga_peer* p) {
if (p->pump) uasync_call_soon_cancel(p->owner->inst->ua, p->pump);
etcp_router_cancel_send_ready(p->owner->inst, p->group, p->peer, ETCP_RT_ID_GROUP_AVATAR, &p->waiter);
ga_release(p, 1); ga_release(p, 0); u_free(p);
}
static void ga_ready(struct ll_queue* queue, void* arg) { (void)queue; ga_schedule(arg); }
static void ga_schedule(struct ga_peer* p) {
if (p->owner->closing || p->pump || p->waiter.internal || p->waiter.call_soon_id) return;
if (!p->state_pending && !p->get_pending && !p->reply_pending && !p->tx) return;
p->pump = uasync_call_soon(p->owner->inst->ua, p, ga_pump);
if (!p->pump) DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: send scheduling failed");
}
static void ga_state(uint8_t* bytes, int kind, uint64_t version, uint16_t size, const uint8_t hash[32]) {
bytes[0] = GA_WIRE_VERSION; bytes[1] = (uint8_t)kind; mf_put64(bytes + 2, version);
bytes[10] = (uint8_t)size; bytes[11] = (uint8_t)(size >> 8); memcpy(bytes + 12, hash, 32);
}
static void ga_pump(void* arg) {
struct ga_peer* p = arg; p->pump = NULL; struct group_avatar* ctx = p->owner;
if (!ga_allowed(ctx->inst, p->group, p->peer)) return;
struct ETCP_ROUTER_CONN* router = etcp_router_conn_get(ctx->inst, p->group, p->peer, ETCP_RT_ID_GROUP_AVATAR);
if (!router) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: router allocation failed"); return; }
queue_set_threshold(router->send_q, 0, 0);
for (unsigned step = 0; step < 4; step++) {
if (!p->state_pending && !p->get_pending && !p->reply_pending && !p->tx) break;
if (!etcp_router_send_q_has_room(ctx->inst, p->group, p->peer, ETCP_RT_ID_GROUP_AVATAR)) {
etcp_router_on_send_ready(ctx->inst, p->group, p->peer, ETCP_RT_ID_GROUP_AVATAR, &p->waiter, ga_ready, p); return;
}
uint8_t packet[GA_DATA_HEADER + GROUP_AVATAR_BLOCK]; size_t length = GA_STATE_BYTES; int kind; uint16_t block = 0;
struct group_avatar_record current;
if (p->state_pending) {
if (group_avatar_get(ctx->inst, p->group, &current) < 0) return;
kind = GA_STATE; ga_state(packet, kind, current.version, current.size, current.hash);
} else if (p->get_pending) { kind = GA_GET; ga_state(packet, kind, p->remote_version, 0, p->remote_hash); }
else {
if (!p->tx) {
if (ctx->tx_count >= GA_STREAMS) return;
p->tx = u_malloc(sizeof(*p->tx));
if (!p->tx) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: send snapshot allocation failed"); return; }
ctx->tx_count++; p->tx_offset = 0; p->tx_at = get_time_tb(); p->reply_pending = 0;
if (group_avatar_get(ctx->inst, p->group, p->tx) < 0) { ga_release(p, 1); return; }
if (!p->tx->version) { ga_release(p, 1); p->state_pending = 1; continue; }
}
kind = GA_DATA; block = p->tx->size - p->tx_offset;
if (block > GROUP_AVATAR_BLOCK) block = GROUP_AVATAR_BLOCK;
ga_state(packet, kind, p->tx->version, p->tx->size, p->tx->hash);
memcpy(packet + 44, p->tx->signature, 64);
packet[108] = (uint8_t)p->tx_offset; packet[109] = (uint8_t)(p->tx_offset >> 8);
packet[110] = (uint8_t)block; packet[111] = (uint8_t)(block >> 8);
memcpy(packet + GA_DATA_HEADER, p->tx->data + p->tx_offset, block); length = GA_DATA_HEADER + block;
}
if (etcp_router_conn_send(router, packet, length, ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED)) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT, "group avatar: send failed group=%llu peer=%016llx kind=%d",
(unsigned long long)p->group, (unsigned long long)p->peer, kind); return;
}
if (kind == GA_STATE) { p->state_pending = 0; p->advertised_at = get_time_tb(); }
else if (kind == GA_GET) { p->get_pending = 0; p->requested_at = get_time_tb(); }
else {
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT, "group avatar: block sent group=%llu peer=%016llx version=%llu offset=%u bytes=%u total=%u",
(unsigned long long)p->group, (unsigned long long)p->peer, (unsigned long long)p->tx->version,
p->tx_offset, block, p->tx->size);
p->tx_offset += block; p->tx_at = get_time_tb();
if (p->tx_offset == p->tx->size) ga_release(p, 1);
}
}
ga_schedule(p);
}
static int ga_receive_ready(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, const uint8_t* data, size_t size) {
struct group_avatar* ctx = ga_context(inst);
if (!ctx || size < GA_DATA_HEADER || data[0] != GA_WIRE_VERSION || data[1] != GA_DATA || !ga_allowed(inst, group, peer)) return 1;
struct ga_peer* p = ga_find(ctx, group, peer, 0);
return (p && p->rx) || ctx->rx_count < GA_STREAMS;
}
static void ga_receive(struct ETCP_CONN* conn, struct ll_entry* entry) {
struct group_avatar* ctx = conn ? ga_context(conn->instance) : NULL;
if (!ctx || ctx->closing || !entry || !entry->dgram || entry->len < ROUTER_SVC_PAYLOAD_OFF) goto out;
uint64_t group, peer; memcpy(&group, entry->dgram + ROUTER_SVC_GROUP_OFF, 8); memcpy(&peer, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
struct ga_peer* p = ga_find(ctx, group, peer, 0);
if (entry->len == ROUTER_SVC_PAYLOAD_OFF) {
if (p) { ga_release(p, 1); ga_release(p, 0); p->requested_at = 0; p->state_pending = 1; ga_schedule(p); }
goto out;
}
if ((entry->dgram[ROUTER_SVC_FLAGS_OFF] & (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED)) !=
(ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED) || !ga_allowed(ctx->inst, group, peer)) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT, "group avatar: unauthorized packet group=%llu peer=%016llx",
(unsigned long long)group, (unsigned long long)peer); goto out;
}
const uint8_t* bytes = entry->dgram + ROUTER_SVC_PAYLOAD_OFF; size_t length = entry->len - ROUTER_SVC_PAYLOAD_OFF;
if (length < GA_STATE_BYTES || bytes[0] != GA_WIRE_VERSION || bytes[1] < GA_STATE || bytes[1] > GA_DATA) goto malformed;
uint64_t version = mf_get64(bytes + 2); uint16_t size = bytes[10] | ((uint16_t)bytes[11] << 8);
if (version > INT64_MAX || size > GROUP_AVATAR_MAX) goto malformed;
if (!p) p = ga_find(ctx, group, peer, 1);
if (!p) goto out;
if (bytes[1] != GA_DATA) {
if (length != GA_STATE_BYTES) goto malformed;
struct group_avatar_record local; if (group_avatar_get(ctx->inst, group, &local) < 0) goto out;
if (bytes[1] == GA_GET) {
if (!p->tx) p->reply_pending = 1;
} else {
p->remote_version = version; memcpy(p->remote_hash, bytes + 12, 32);
int newer = ga_compare(version, p->remote_hash, local.version, local.hash);
if (newer > 0 && (!p->requested_at || get_time_tb() - p->requested_at >= GA_RETRY_TB)) p->get_pending = 1;
if (newer < 0) p->state_pending = 1;
if (!newer) p->requested_at = 0;
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT, "group avatar: state group=%llu peer=%016llx remote=%llu local=%llu request=%d",
(unsigned long long)group, (unsigned long long)peer, (unsigned long long)version,
(unsigned long long)local.version, p->get_pending);
}
ga_schedule(p); goto out;
}
if (length < GA_DATA_HEADER || !version) goto malformed;
uint16_t offset = bytes[108] | ((uint16_t)bytes[109] << 8), block = bytes[110] | ((uint16_t)bytes[111] << 8);
if (offset > size || block > GROUP_AVATAR_BLOCK || block > size - offset || length != GA_DATA_HEADER + block ||
(size && !block) || (offset + block < size && block != GROUP_AVATAR_BLOCK)) goto malformed;
if (!offset) {
ga_release(p, 0);
if (ctx->rx_count >= GA_STREAMS) { DEBUG_WARN(DEBUG_CATEGORY_CHAT, "group avatar: receive limit reached"); goto out; }
p->rx = u_calloc(1, sizeof(*p->rx));
if (!p->rx) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: receive allocation failed"); goto out; }
ctx->rx_count++; p->rx_offset = 0; p->rx->version = version; p->rx->size = size;
memcpy(p->rx->hash, bytes + 12, 32); memcpy(p->rx->signature, bytes + 44, 64);
}
if (!p->rx || p->rx_offset != offset || p->rx->version != version || p->rx->size != size ||
memcmp(p->rx->hash, bytes + 12, 32) || memcmp(p->rx->signature, bytes + 44, 64)) goto malformed;
memcpy(p->rx->data + offset, bytes + GA_DATA_HEADER, block); p->rx_offset += block; p->rx_at = get_time_tb();
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT, "group avatar: block received group=%llu peer=%016llx version=%llu offset=%u bytes=%u total=%u",
(unsigned long long)group, (unsigned long long)peer, (unsigned long long)version, offset, block, size);
if (p->rx_offset == size) {
int applied = group_avatar_apply(ctx->inst, group, p->rx); ga_release(p, 0);
p->requested_at = applied < 0 ? get_time_tb() : 0;
p->state_pending = 1; ga_schedule(p);
}
goto out;
malformed:
DEBUG_WARN(DEBUG_CATEGORY_CHAT, "group avatar: malformed/mixed blocks group=%llu peer=%016llx bytes=%zu",
(unsigned long long)group, (unsigned long long)peer, length);
if (p) DEBUG_WARN(DEBUG_CATEGORY_CHAT, "group avatar: receive state rx=%d offset=%u version=%llu requested=%llu",
p->rx != NULL, p->rx_offset, (unsigned long long)(p->rx ? p->rx->version : 0), (unsigned long long)p->requested_at);
if (p) { ga_release(p, 0); p->requested_at = get_time_tb(); }
out:
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
}
static void ga_tick(void* arg) {
struct group_avatar* ctx = arg; ctx->timer = NULL; uint64_t now = get_time_tb();
for (struct ga_peer* p = ctx->peers; p; p = p->next) p->seen = 0;
for (struct ll_entry* ge = ctx->inst->topo_groups->group_list->head; ge; ge = ge->next) {
struct TOPO_GROUP* group = (struct TOPO_GROUP*)ge;
if (group->group_type != TOPO_GROUP_TYPE_CHAT || !group->senders_list) continue;
for (struct ll_entry* e = group->senders_list->head; e; e = e->next) {
struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data;
if (!item || !item->conn || !item->ready_notified || item->transport_down || !ga_allowed(ctx->inst, group->group_id, item->node_id)) continue;
struct ga_peer* p = ga_find(ctx, group->group_id, item->node_id, 1); if (!p) continue;
p->seen = 1;
if (p->epoch != item->local_epoch) {
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT, "group avatar: session epoch reset group=%llu peer=%016llx old=%llu new=%llu tx=%d rx=%d",
(unsigned long long)p->group, (unsigned long long)p->peer, (unsigned long long)p->epoch,
(unsigned long long)item->local_epoch, p->tx != NULL, p->rx != NULL);
ga_release(p, 1); ga_release(p, 0); p->epoch = item->local_epoch; p->requested_at = 0; p->state_pending = 1;
}
if (now - p->advertised_at >= GA_EXCHANGE_TB) p->state_pending = 1;
if (p->requested_at && now - p->requested_at >= GA_RETRY_TB) { p->get_pending = 1; p->requested_at = 0; }
if ((p->rx && now - p->rx_at >= GA_RETRY_TB) || (p->tx && now - p->tx_at >= GA_RETRY_TB)) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT, "group avatar: incomplete transfer expired group=%llu peer=%016llx",
(unsigned long long)p->group, (unsigned long long)p->peer); ga_release(p, 1); ga_release(p, 0); p->state_pending = 1;
}
ga_schedule(p);
}
}
struct ga_peer** link = &ctx->peers;
while (*link) { struct ga_peer* p = *link; if (p->seen) link = &p->next; else { *link = p->next; ga_free(p); } }
ctx->timer = uasync_set_timeout(ctx->inst->ua, 10000, ctx, ga_tick, "group_avatar");
if (!ctx->timer) DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: timer allocation failed");
}
static int ga_schema(sqlite3* db) {
const char* names[] = {"avatar_version", "avatar_data", "avatar_hash", "avatar_signature"}; int present[4] = {0};
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(db, "PRAGMA table_info(channels)", -1, &st, NULL) != SQLITE_OK) goto error;
while (sqlite3_step(st) == SQLITE_ROW) for (unsigned i = 0; i < 4; i++)
if (!strcmp((const char*)sqlite3_column_text(st, 1), names[i])) present[i] = 1;
sqlite3_finalize(st);
if (sqlite3_exec(db, "SAVEPOINT group_avatar_schema", NULL, NULL, NULL) != SQLITE_OK) goto error;
for (unsigned i = 0; i < 4; i++) if (!present[i]) {
char sql[128]; snprintf(sql, sizeof(sql), "ALTER TABLE channels ADD COLUMN %s %s", names[i], i ? "BLOB" : "INTEGER NOT NULL DEFAULT 0");
if (sqlite3_exec(db, sql, NULL, NULL, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: cannot add schema column=%s: %s", names[i], sqlite3_errmsg(db));
if (sqlite3_exec(db, "ROLLBACK TO group_avatar_schema; RELEASE group_avatar_schema", NULL, NULL, NULL) != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: schema rollback failed: %s", sqlite3_errmsg(db));
return -1;
}
DEBUG_INFO(DEBUG_CATEGORY_CHAT, "group avatar: added schema column=%s", names[i]);
}
if (sqlite3_exec(db, "RELEASE group_avatar_schema", NULL, NULL, NULL) == SQLITE_OK) return 0;
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: schema commit failed: %s", sqlite3_errmsg(db));
if (sqlite3_exec(db, "ROLLBACK TO group_avatar_schema; RELEASE group_avatar_schema", NULL, NULL, NULL) != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: schema rollback after commit error failed: %s", sqlite3_errmsg(db));
return -1;
error:
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: schema initialization failed: %s", sqlite3_errmsg(db)); return -1;
}
int group_avatar_init(struct UTUN_INSTANCE* inst) {
if (!inst || !CC(inst) || !CC(inst)->db) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: core unavailable"); return -1; }
if (ga_context(inst)) return 0;
if (ga_schema(CC(inst)->db) < 0) return -1;
struct group_avatar* ctx = u_calloc(1, sizeof(*ctx));
if (!ctx) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: context allocation failed"); return -1; }
ctx->inst = inst; CC(inst)->group_avatar = ctx;
if (etcp_router_bind(inst, ETCP_RT_ID_GROUP_AVATAR, ga_receive)) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: router registration failed"); CC(inst)->group_avatar = NULL; u_free(ctx); return -1;
}
etcp_router_set_receive_ready(inst, ETCP_RT_ID_GROUP_AVATAR, ga_receive_ready);
ctx->timer = uasync_set_timeout(inst->ua, 10000, ctx, ga_tick, "group_avatar");
if (!ctx->timer) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "group avatar: initial timer failed"); group_avatar_destroy(inst); return -1; }
DEBUG_INFO(DEBUG_CATEGORY_CHAT, "group avatar: initialized max_bytes=%d block_bytes=%d", GROUP_AVATAR_MAX, GROUP_AVATAR_BLOCK); return 0;
}
int group_avatar_schema_init(sqlite3* db) { return ga_schema(db); }
void group_avatar_cancel_group(struct UTUN_INSTANCE* inst, uint64_t group) {
struct group_avatar* ctx = ga_context(inst); if (!ctx) return;
struct ga_peer** link = &ctx->peers;
while (*link) { struct ga_peer* p = *link; if (p->group != group) link = &p->next; else { *link = p->next; ga_free(p); } }
}
void group_avatar_destroy(struct UTUN_INSTANCE* inst) {
struct group_avatar* ctx = ga_context(inst); if (!ctx) return;
ctx->closing = 1; if (ctx->timer) uasync_cancel_timeout(inst->ua, ctx->timer);
etcp_router_set_receive_ready(inst, ETCP_RT_ID_GROUP_AVATAR, NULL);
etcp_router_unbind(inst, ETCP_RT_ID_GROUP_AVATAR);
while (ctx->peers) { struct ga_peer* p = ctx->peers; ctx->peers = p->next; ga_free(p); }
CC(inst)->group_avatar = NULL; u_free(ctx); DEBUG_INFO(DEBUG_CATEGORY_CHAT, "group avatar: stopped");
}
Reference in new issue View Git Blame Copy Permalink