utun2/src/chat/group_avatar.c

#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) || topo_node_sqlite_channel_get(CC(inst)->db, ch, NULL, 0, NULL, NULL, pub, NULL) < 0 ||
        sc_ed25519_verify(pub, bytes, ga_message(group, r, bytes), r->signature) != SC_OK) {
        DEBUG_WARN(DEBUG_CATEGORY_CHAT, "group avatar: hash/signature rejected group=%llu version=%llu bytes=%u",
            (unsigned long long)group, (unsigned long long)r->version, r->size); 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 || 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 || 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; 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) { 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) {
                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) {
            sqlite3_exec(db, "ROLLBACK TO group_avatar_schema; RELEASE group_avatar_schema", NULL, NULL, NULL); goto error;
        }
    }
    if (sqlite3_exec(db, "RELEASE group_avatar_schema", NULL, NULL, NULL) == SQLITE_OK) return 0;
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;
}

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");
}