utun2/src/member_files/member_files.c

#include "private.h"
#include "../utun_instance.h"
#include "../chat/chat_event.h"
#include "../chat/chat_setting.h"
#include "../routing_layer/etcp_router.h"
#include "../routing_layer/conn_mgr.h"
#include "../routing_layer/topo_group.h"
#include "../transport_layer/etcp.h"
#include "../../lib/mem.h"
#include "../../lib/debug_config.h"
#include "../../lib/json_flat.h"
#include <openssl/rand.h>
#include <openssl/evp.h>
#include <string.h>
#include <stdio.h>

#define MF_MAX_PENDING 16
#define MF_MAX_JOBS 32
#define MF_REQUEST_TB 1200000ULL

struct mf_pending {
    struct mf_pending* next;
    struct member_files* files;
    uint64_t group, server, started, sent;
    struct mf_message request;
    mf_response_fn callback;
    void* arg;
    void* pump;
    struct CONN_MGR_HANDLE* cm;
    struct queue_waiter_handle waiter;
};
struct mf_job {
    struct mf_job* next;
    struct member_files* files;
    uint64_t group, peer, created;
    struct mf_message request, response;
    struct queue_waiter_handle waiter;
    int state; /* 0 queued, 1 worker, 2 reply; command=0 — sweep */
};
struct mf_local {
    struct mf_local* next;
    struct member_files* files;
    uint64_t group;
    struct mf_message message;
    void* task;
};

static void mf_server_pump(struct member_files* f);
static void mf_client_ready(struct ll_queue* q, void* arg);
static void mf_dispatch(struct member_files* f, uint64_t group, uint64_t peer, const struct mf_message* message);

static void mf_local_free(struct mf_local* local) {
    struct member_files* f = local->files;
    if (local->task) uasync_call_soon_cancel(f->inst->ua, local->task);
    struct mf_local** link = &f->locals;
    while (*link && *link != local) link = &(*link)->next;
    if (*link) *link = local->next;
    f->local_count--; u_free(local);
}

/* Локальная доставка всегда отложена: ответ не освобождает pending внутри send. */
static void mf_local_deliver(void* arg) {
    struct mf_local* local = arg; local->task = NULL;
    struct member_files* f = local->files;
    struct mf_local** link = &f->locals;
    while (*link && *link != local) link = &(*link)->next;
    if (*link) *link = local->next;
    f->local_count--;
    if (!f->closing) mf_dispatch(f, local->group, f->inst->node_id, &local->message);
    u_free(local);
}

int member_files_member(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t node, int storage) {
    struct TOPO_GROUP* g = inst && inst->topo_groups ? topo_groups_find(inst->topo_groups, group) : NULL;
    if (!g || g->group_type != TOPO_GROUP_TYPE_CHAT || !node || !inst->topo_sqlite_db) {
        DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: ACL missing CHAT group=%llu node=%llu role=%d",
                    (unsigned long long)group, (unsigned long long)node, storage); return 0;
    }
    char sql[256];
    snprintf(sql, sizeof(sql), "SELECT adm_tags,source,adm_tags_sig,deleted FROM \"peers_%llu\" WHERE node_id=?",
             (unsigned long long)group);
    sqlite3_stmt* st = NULL;
    if (sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL) != SQLITE_OK) {
        DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: membership query failed group=%llu: %s",
                    (unsigned long long)group, sqlite3_errmsg(inst->topo_sqlite_db)); return 0;
    }
    sqlite3_bind_int64(st, 1, (sqlite3_int64)node);
    int rc = sqlite3_step(st);
    char role[16]; const char* tags = (const char*)sqlite3_column_text(st, 0);
    int storage_role = tags && json_flat_get(tags, "storage", role, sizeof(role)) == 0 && !strcmp(role, "yes");
    int allowed = rc == SQLITE_ROW && !sqlite3_column_int(st, 1) && !sqlite3_column_int(st, 3) &&
                  (!storage || (storage_role && sqlite3_column_bytes(st, 2) == 64));
    DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
        "member_files: ACL allowed=%d group=%llu node=%llu require_storage=%d row=%d storage=%d source=%d signature=%d deleted=%d",
        allowed, (unsigned long long)group, (unsigned long long)node, storage, rc == SQLITE_ROW,
        storage_role, sqlite3_column_int(st, 1), sqlite3_column_bytes(st, 2), sqlite3_column_int(st, 3));
    if (rc != SQLITE_ROW && rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: membership step failed rc=%d", rc);
    sqlite3_finalize(st); return allowed;
}

static int mf_server_allowed(struct member_files* f, uint64_t group, uint64_t peer, const struct mf_message* q) {
    int enabled = chat_setting_get_int(f->inst, "member_files_enabled", 1);
    int allowed = enabled && member_files_member(f->inst, group, f->inst->node_id, 1) &&
                  member_files_member(f->inst, group, peer, 0) && (!q->owner || member_files_member(f->inst, group, q->owner, 0)) &&
                  (q->command < MF_BEGIN_WRITE || (q->owner && q->owner == peer));
    DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: server access id=%llu cmd=%s enabled=%d allowed=%d owner_is_peer=%d",
        (unsigned long long)q->request_id, mf_command_text(q->command), enabled, allowed, q->owner == peer);
    return allowed;
}

/* Router подтверждает доставку отдельно; здесь отправка всегда соблюдает порог очереди. */
static int mf_send(struct member_files* f, uint64_t group, uint64_t peer, const struct mf_message* m) {
    if (peer == f->inst->node_id) {
        if (f->local_count >= MF_MAX_JOBS) return 1;
        struct mf_local* local = u_calloc(1, sizeof(*local));
        if (!local) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: local delivery allocation failed"); return -1; }
        local->files = f; local->group = group; local->message = *m;
        local->task = uasync_call_soon(f->inst->ua, local, mf_local_deliver);
        if (!local->task) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: local delivery scheduling failed"); u_free(local); return -1; }
        local->next = f->locals; f->locals = local; f->local_count++;
        DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: local queued id=%llu cmd=%s kind=%u count=%u",
                    (unsigned long long)m->request_id, mf_command_text(m->command), m->kind, f->local_count);
        return 0;
    }
    struct ETCP_ROUTER_CONN* rc = etcp_router_conn_get(f->inst, group, peer, ETCP_RT_ID_MEMBER_FILES);
    if (!rc) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: router allocation failed"); return -1; }
    queue_set_threshold(rc->send_q, 0, 0); etcp_router_set_max_inflight(rc, 16);
    if (!etcp_router_send_q_has_room(f->inst, group, peer, ETCP_RT_ID_MEMBER_FILES)) {
        DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: tx waiting id=%llu cmd=%s send_q=%u inflight=%u sync=%u no_route=%u",
            (unsigned long long)m->request_id, mf_command_text(m->command), rc->send_q->count, rc->inflight_q->count,
            rc->peer_sync_done, rc->no_route);
        return 1;
    }
    uint8_t wire[MF_WIRE_MAX];
    int n = mf_encode(m, wire, sizeof(wire));
    if (n < 0) return -1;
    uint32_t seq = rc->tx_seq;
    int result = etcp_router_conn_send(rc, wire, (size_t)n, ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED);
    DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
        "member_files: tx id=%llu kind=%u cmd=%s group=%llx peer=%llx owner=%llx status=%s bytes=%d seq=%u accepted=%d",
        (unsigned long long)m->request_id, m->kind, mf_command_text(m->command), (unsigned long long)group,
        (unsigned long long)peer, (unsigned long long)m->owner, mf_status_text(m->status), n, seq, result == 0);
    if (result) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: send failed group=%llu peer=%llu cmd=%u rc=%d",
                           (unsigned long long)group, (unsigned long long)peer, m->command, result);
    return result ? -1 : 0;
}

static void mf_pending_finish(struct mf_pending* p, const struct mf_message* response, int status) {
    struct member_files* f = p->files;
    if (p->pump) uasync_call_soon_cancel(f->inst->ua, p->pump);
    etcp_router_cancel_send_ready(f->inst, p->group, p->server, ETCP_RT_ID_MEMBER_FILES, &p->waiter);
    if (p->cm) conn_mgr_close(p->cm);
    struct mf_pending** link = &f->pending;
    while (*link && *link != p) link = &(*link)->next;
    if (*link) *link = p->next;
    f->pending_count--;
    struct mf_message result;
    if (!response) { result = p->request; result.kind = MF_RESPONSE; result.status = (uint16_t)status; result.data_len = 0; response = &result; }
    mf_response_fn cb = p->callback; void* arg = p->arg;
    DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: request done id=%llu cmd=%s status=%s pending=%u elapsed_ms=%llu",
        (unsigned long long)p->request.request_id, mf_command_text(p->request.command), mf_status_text(response->status),
        f->pending_count, (unsigned long long)((get_time_tb() - p->started) / 10));
    u_free(p); cb(arg, response);
}

static void mf_pending_pump(void* arg) {
    struct mf_pending* p = arg; p->pump = NULL;
    if (p->files->closing) return;
    int rc = mf_send(p->files, p->group, p->server, &p->request);
    if (!rc) p->sent = get_time_tb();
    else if (rc > 0) etcp_router_on_send_ready(p->files->inst, p->group, p->server, ETCP_RT_ID_MEMBER_FILES,
                                              &p->waiter, mf_client_ready, p);
}

static void mf_pending_wake(struct mf_pending* p) {
    if (p->pump || p->files->closing) return;
    p->pump = uasync_call_soon(p->files->inst->ua, p, mf_pending_pump);
    if (!p->pump) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: client pump allocation failed");
}

static void mf_client_ready(struct ll_queue* q, void* arg) { (void)q; mf_pending_wake(arg); }

static void mf_connected(struct CONN_MGR_HANDLE* h, uint64_t peer, uint64_t group, enum conn_mgr_event event, void* arg) {
    (void)h; (void)peer; (void)group;
    struct mf_pending* p = arg;
    DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: CM event=%d request=%llu", event, (unsigned long long)p->request.request_id);
    if (event == CONN_EVENT_UP) mf_pending_wake(p);
}

struct mf_pending* member_files_request(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t server,
                                       const struct mf_message* request, mf_response_fn cb, void* arg) {
    struct member_files* f = inst ? inst->member_files : NULL;
    if (request && (request->command < MF_LIST || request->command > MF_REMOVE || request->status ||
        request->data_len > MF_BLOCK_SIZE || !mf_path_valid(request->path, request->command == MF_LIST || request->command == MF_STAT) ||
        (request->command == MF_RENAME && !mf_path_valid(request->target, 0)))) {
        DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: invalid client command=%u status=%u data=%u",
                   request->command, request->status, request->data_len); return NULL;
    }
    if (!f || f->closing || !request || !cb || f->pending_count >= MF_MAX_PENDING ||
        !member_files_member(inst, group, inst->node_id, 0) || !member_files_member(inst, group, server, 1)) {
        DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: client request denied group=%llu server=%llu",
                   (unsigned long long)group, (unsigned long long)server); return NULL;
    }
    struct mf_pending* p = u_calloc(1, sizeof(*p));
    if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: client allocation failed"); return NULL; }
    p->files = f; p->group = group; p->server = server; p->request = *request; p->request.kind = MF_REQUEST;
    p->callback = cb; p->arg = arg; p->started = get_time_tb();
    if (!p->request.request_id && (RAND_bytes((uint8_t*)&p->request.request_id, 8) != 1 || !p->request.request_id)) {
        DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: request random failed"); u_free(p); return NULL;
    }
    for (struct mf_pending* other = f->pending; other; other = other->next) if (other->request.request_id == p->request.request_id) {
        DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: duplicate local request id"); u_free(p); return NULL;
    }
    p->next = f->pending; f->pending = p; f->pending_count++;
    DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
        "member_files: request id=%llu cmd=%s group=%llx server=%llx owner=%llx path=%s offset=%llu size=%llu pending=%u",
        (unsigned long long)p->request.request_id, mf_command_text(p->request.command), (unsigned long long)group,
        (unsigned long long)server, (unsigned long long)p->request.owner, p->request.path,
        (unsigned long long)p->request.offset, (unsigned long long)p->request.size, f->pending_count);
    if (server != inst->node_id && conn_mgr_open(inst, group, server, mf_connected, p, &p->cm) != 0) {
        DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: CM launch failed server=%llu", (unsigned long long)server);
        if (p->pump) uasync_call_soon_cancel(inst->ua, p->pump);
        f->pending = p->next; f->pending_count--; u_free(p); return NULL;
    }
    mf_pending_wake(p);
    return p;
}

void member_files_cancel(struct mf_pending* p) { if (p) mf_pending_finish(p, NULL, MF_CANCELLED); }

void member_files_cancel_request_id(struct UTUN_INSTANCE* inst, uint64_t id) {
    struct member_files* f = inst ? inst->member_files : NULL;
    if (f) for (struct mf_pending* p = f->pending; p; p = p->next) if (p->request.request_id == id) {
        member_files_cancel(p); break;
    }
}

static void mf_job_free(struct mf_job* j) {
    struct member_files* f = j->files;
    etcp_router_cancel_send_ready(f->inst, j->group, j->peer, ETCP_RT_ID_MEMBER_FILES, &j->waiter);
    struct mf_job** link = &f->jobs;
    while (*link && *link != j) link = &(*link)->next;
    if (*link) *link = j->next;
    f->job_count--; u_free(j);
}

static void mf_server_deferred(void* arg) {
    struct member_files* f = arg; f->pump = NULL; mf_server_pump(f);
}
static void mf_server_ready(struct ll_queue* q, void* arg) {
    (void)q;
    struct member_files* f = arg;
    if (!f->pump && !f->closing) {
        f->pump = uasync_call_soon(f->inst->ua, f, mf_server_deferred);
        if (!f->pump) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: server pump allocation failed");
    }
}

static void mf_server_work(void* arg) {
    struct mf_job* j = arg;
    if (j->request.command) DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: disk start id=%llu cmd=%s group=%llx peer=%llx path=%s",
        (unsigned long long)j->request.request_id, mf_command_text(j->request.command), (unsigned long long)j->group,
        (unsigned long long)j->peer, j->request.path);
    if (j->request.command) mf_storage_execute(j->files->storage, j->group, j->peer, &j->request, &j->response);
    else mf_storage_sweep(j->files->storage, get_time_tb());
}

static void mf_server_done(void* arg, int error) {
    struct mf_job* j = arg;
    struct member_files* f = j->files; f->worker = NULL;
    if (f->closing || !j->request.command) { mf_job_free(j); if (!f->closing) mf_server_pump(f); return; }
    if (error) { j->response = j->request; j->response.kind = MF_RESPONSE; j->response.status = MF_BUSY; j->response.data_len = 0; }
    DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: disk done id=%llu cmd=%s status=%s error=%d jobs=%u",
        (unsigned long long)j->request.request_id, mf_command_text(j->request.command), mf_status_text(j->response.status), error, f->job_count);
    j->state = 2; mf_server_pump(f);
}

/* Виртуальный корень — действующие участники, в том числе ещё без файлов на диске. */
static void mf_roots(struct mf_job* j) {
    struct mf_message* r = &j->response;
    *r = j->request; r->kind = MF_RESPONSE; r->type = MF_DIRECTORY; r->data_len = 0; r->more = 0;
    if (*r->path || r->command != MF_LIST) { r->status = MF_INVALID_PATH; return; }
    char sql[256];
    snprintf(sql, sizeof(sql), "SELECT node_id,join_ts,update_ts,adm_tags FROM \"peers_%llu\" WHERE deleted=0 AND source=0 ORDER BY node_id",
             (unsigned long long)j->group);
    sqlite3_stmt* st = NULL; EVP_MD_CTX* hash = EVP_MD_CTX_new();
    if (!hash || EVP_DigestInit_ex(hash, EVP_sha256(), NULL) != 1 ||
        sqlite3_prepare_v2(j->files->inst->topo_sqlite_db, sql, -1, &st, NULL) != SQLITE_OK) {
        DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: virtual root query/hash failed"); r->status = MF_IO_ERROR; goto done;
    }
    uint64_t seen = 0; int rc;
    while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
        uint8_t metadata[24];
        uint64_t id = (uint64_t)sqlite3_column_int64(st, 0);
        mf_put64(metadata, id); mf_put64(metadata + 8, (uint64_t)sqlite3_column_int64(st, 1));
        mf_put64(metadata + 16, (uint64_t)sqlite3_column_int64(st, 2));
        const char* tags = (const char*)sqlite3_column_text(st, 3);
        if (EVP_DigestUpdate(hash, metadata, sizeof(metadata)) != 1 ||
            (tags && EVP_DigestUpdate(hash, tags, strlen(tags) + 1) != 1)) { r->status = MF_IO_ERROR; goto done; }
        if (seen++ < j->request.cursor || r->more) continue;
        char name[32]; int n = snprintf(name, sizeof(name), "%llu", (unsigned long long)id);
        if (r->data_len + MF_ENTRY_HEADER + (unsigned)n > MF_BLOCK_SIZE) { r->more = 1; continue; }
        uint8_t* p = r->data + r->data_len;
        memset(p, 0, MF_ENTRY_HEADER); p[0] = (uint8_t)n; p[2] = MF_DIRECTORY;
        memcpy(p + MF_ENTRY_HEADER, name, (size_t)n); r->data_len += MF_ENTRY_HEADER + (uint32_t)n; r->cursor++;
    }
    uint8_t digest[32]; unsigned n;
    if (rc != SQLITE_DONE || EVP_DigestFinal_ex(hash, digest, &n) != 1 || n != 32) { r->status = MF_IO_ERROR; goto done; }
    memcpy(r->version, digest, 16);
    if ((j->request.cursor && mf_zero(j->request.version, 16)) ||
        (!mf_zero(j->request.version, 16) && memcmp(j->request.version, r->version, 16))) r->status = MF_STALE;
done:
    sqlite3_finalize(st); EVP_MD_CTX_free(hash);
    if (r->status) r->data_len = 0;
}

/* Один disk worker: I/O, sweep и namespace-мутации никогда не пересекаются. */
static void mf_server_pump(struct member_files* f) {
    if (f->closing) return;
    struct mf_job* j = f->jobs;
    while (j) {
        struct mf_job* next = j->next;
        if (j->state == 2 && !j->waiter.internal) {
            if (j->response.status == MF_OK && !mf_server_allowed(f, j->group, j->peer, &j->request)) {
                j->response.status = MF_ACCESS_DENIED; j->response.data_len = 0;
            }
            int result = mf_send(f, j->group, j->peer, &j->response);
            if (result <= 0) mf_job_free(j);
            else etcp_router_on_send_ready(f->inst, j->group, j->peer, ETCP_RT_ID_MEMBER_FILES, &j->waiter, mf_server_ready, f);
        }
        j = next;
    }
    if (f->worker) {
        DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: server waiting for disk id=%llu cmd=%s jobs=%u",
            (unsigned long long)f->worker->request.request_id, mf_command_text(f->worker->request.command), f->job_count);
        return;
    }
    for (j = f->jobs; j; j = j->next) if (!j->state) break;
    if (!j && f->sweep_due && f->job_count < MF_MAX_JOBS) {
        j = u_calloc(1, sizeof(*j));
        if (!j) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: sweep allocation failed"); return; }
        j->files = f; j->next = f->jobs; f->jobs = j; f->job_count++; f->sweep_due = 0;
    }
    if (!j) return;
    if (j->request.command && !mf_server_allowed(f, j->group, j->peer, &j->request)) {
        j->response = j->request; j->response.kind = MF_RESPONSE; j->response.status = MF_ACCESS_DENIED;
        j->response.data_len = 0; j->state = 2; mf_server_pump(f); return;
    }
    j->state = 1; f->worker = j;
    if (j->request.command) DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: disk queued id=%llu cmd=%s jobs=%u",
        (unsigned long long)j->request.request_id, mf_command_text(j->request.command), f->job_count);
    mf_worker_submit(f->disk, mf_server_work, j, mf_server_done, j);
}

static void mf_tick(void* arg) {
    struct member_files* f = arg; f->tick = NULL;
    uint64_t now = get_time_tb();
    for (;;) {
        struct mf_pending* expired = NULL;
        for (struct mf_pending* p = f->pending; p; p = p->next) if (now - p->started >= MF_REQUEST_TB) { expired = p; break; }
        if (!expired) break;
        DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: timeout id=%llu cmd=%u",
                   (unsigned long long)expired->request.request_id, expired->request.command);
        mf_pending_finish(expired, NULL, MF_TIMEOUT);
    }
    for (struct mf_pending* p = f->pending; p; p = p->next)
        if (!p->sent || now - p->sent >= 50000) mf_pending_wake(p);
    struct mf_job* j = f->jobs;
    while (j) {
        struct mf_job* next = j->next;
        if (j != f->worker && j->created && now - j->created >= MF_REQUEST_TB) {
            DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: server reply/queue timeout peer=%llu", (unsigned long long)j->peer); mf_job_free(j);
        }
        j = next;
    }
    f->sweep_due = f->used_storage; mf_server_pump(f);
    f->tick = uasync_set_timeout(f->inst->ua, 10000, f, mf_tick, "member_files");
    if (!f->tick) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: timer allocation failed");
}

/* Общая обработка проверенного сетевого сообщения или команды внутри экземпляра. */
static void mf_dispatch(struct member_files* f, uint64_t group, uint64_t peer, const struct mf_message* message) {
    struct mf_message m = *message;
    DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: rx id=%llu kind=%u cmd=%s group=%llx peer=%llx owner=%llx jobs=%u worker=%p",
        (unsigned long long)m.request_id, m.kind, mf_command_text(m.command), (unsigned long long)group,
        (unsigned long long)peer, (unsigned long long)m.owner, f->job_count, (void*)f->worker);
    if (m.kind == MF_RESPONSE) {
        for (struct mf_pending* p = f->pending; p; p = p->next) {
            if (p->group != group || p->server != peer || p->request.request_id != m.request_id) continue;
            if (p->request.command != m.command || p->request.owner != m.owner || strcmp(p->request.path, m.path)) {
                DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: mismatched response id=%llu", (unsigned long long)m.request_id); return;
            }
            if (!member_files_member(f->inst, group, peer, 1)) { m.status = MF_ACCESS_DENIED; m.data_len = 0; }
            mf_pending_finish(p, &m, 0); break;
        }
        return;
    }
    for (struct mf_job* other = f->jobs; other; other = other->next) if (other->group == group && other->peer == peer &&
        other->request.request_id == m.request_id) return;
    if (f->job_count >= MF_MAX_JOBS) {
        DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: server queue full peer=%llu", (unsigned long long)peer); return;
    }
    struct mf_job* j = u_calloc(1, sizeof(*j));
    if (!j) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: server job allocation failed"); return; }
    j->files = f; j->group = group; j->peer = peer; j->created = get_time_tb(); j->request = m;
    struct mf_job** tail = &f->jobs; while (*tail) tail = &(*tail)->next;
    *tail = j; f->job_count++;
    if (!mf_server_allowed(f, group, peer, &m)) {
        j->response = m; j->response.kind = MF_RESPONSE; j->response.status = MF_ACCESS_DENIED; j->response.data_len = 0; j->state = 2;
        DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: ACL denied group=%llu peer=%llu owner=%llu cmd=%u",
                   (unsigned long long)group, (unsigned long long)peer, (unsigned long long)m.owner, m.command);
    } else if (!m.owner) { mf_roots(j); j->state = 2; }
    else f->used_storage = 1;
    mf_server_pump(f);
}

static void mf_recv(struct ETCP_CONN* conn, struct ll_entry* e) {
    struct member_files* f = conn ? conn->instance->member_files : NULL;
    if (!f || f->closing || !e || !e->dgram || e->len <= ROUTER_SVC_PAYLOAD_OFF) {
        DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: recv discarded context=%p closing=%d bytes=%u",
                    (void*)f, f ? f->closing : 0, e ? e->len : 0);
        goto out;
    }
    uint64_t group, peer;
    memcpy(&group, e->dgram + ROUTER_SVC_GROUP_OFF, 8); memcpy(&peer, e->dgram + ROUTER_SVC_SRC_OFF, 8);
    if ((e->dgram[ROUTER_SVC_FLAGS_OFF] & (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED)) !=
        (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED)) {
        DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: unsigned/unencrypted packet peer=%llu", (unsigned long long)peer); goto out;
    }
    struct mf_message m;
    if (!mf_decode(e->dgram + ROUTER_SVC_PAYLOAD_OFF, e->len - ROUTER_SVC_PAYLOAD_OFF, &m)) mf_dispatch(f, group, peer, &m);
out:
    if (e) { queue_dgram_free(e); queue_entry_free(e); }
}

int member_files_init(struct UTUN_INSTANCE* inst) {
    if (!inst || inst->member_files) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: invalid init"); return -1; }
    struct member_files* f = u_calloc(1, sizeof(*f));
    if (!f) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: owner allocation failed"); return -1; }
    f->inst = inst;
    f->disk = mf_worker_create(inst->ua); f->cleanup = mf_worker_reserve();
    if (!f->disk || !f->cleanup) {
        mf_worker_destroy(f->disk); mf_worker_unreserve(f->cleanup); u_free(f); return -1;
    }
    char path[2048]; const char* custom = chat_setting_get(inst, "member_files_path");
    if (custom && !*custom) custom = NULL;
    snprintf(path, sizeof(path), "%s%s%s", custom ? custom : inst->config->global.db_path, custom ? "" : "/", custom ? "" : "member_files");
    if (!custom && !*inst->config->global.db_path) snprintf(path, sizeof(path), "member_files");
    f->storage = mf_storage_create(path, (uint64_t)chat_setting_get_int(inst, "member_files_quota_mb", 1024) * 1024 * 1024,
                                    (uint64_t)chat_setting_get_int(inst, "member_files_upload_ttl_sec", 3600) * 10000);
    if (!f->storage || etcp_router_bind(inst, ETCP_RT_ID_MEMBER_FILES, mf_recv)) {
        mf_storage_destroy(f->storage); mf_worker_destroy(f->disk); mf_worker_unreserve(f->cleanup); u_free(f); return -1;
    }
    inst->member_files = f;
    f->tick = uasync_set_timeout(inst->ua, 10000, f, mf_tick, "member_files");
    if (!f->tick) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: init timer failed"); member_files_destroy(inst); return -1; }
    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: initialized root=%s idle=10s max_requests=%d max_jobs=%d", path, MF_MAX_PENDING, MF_MAX_JOBS);
    return 0;
}

void member_files_quiesce(struct UTUN_INSTANCE* inst) {
    struct member_files* f = inst ? inst->member_files : NULL;
    if (!f || f->closing) return;
    f->closing = 1;
    if (f->pump) uasync_call_soon_cancel(inst->ua, f->pump);
    f->pump = NULL;
    if (f->tick) uasync_cancel_timeout(inst->ua, f->tick);
    f->tick = NULL;
    mf_transfers_quiesce(f, 0);
    while (f->pending) member_files_cancel(f->pending);
    while (f->locals) mf_local_free(f->locals);
    struct mf_job* j = f->jobs;
    while (j) { struct mf_job* next = j->next; if (j != f->worker) mf_job_free(j); j = next; }
    etcp_router_unbind(inst, ETCP_RT_ID_MEMBER_FILES);
    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: quiesce workers=%d transfers=%u", f->worker != NULL, f->transfer_count);
}

static void mf_storage_cleanup(void* arg) { mf_storage_destroy(arg); }
static void mf_cleanup_done(void* arg, int error) { (void)arg; (void)error; }

void member_files_destroy(struct UTUN_INSTANCE* inst) {
    struct member_files* f = inst ? inst->member_files : NULL;
    if (!f) return;
    member_files_quiesce(inst);
    mf_worker_submit_reserved(f->disk, f->cleanup, mf_storage_cleanup, f->storage, mf_cleanup_done, NULL);
    f->cleanup = NULL;
    mf_worker_destroy(f->disk);
    u_free(f); inst->member_files = NULL;
}

void member_files_cancel_group(struct UTUN_INSTANCE* inst, uint64_t group) {
    struct member_files* f = inst ? inst->member_files : NULL;
    if (!f) return;
    mf_transfers_quiesce(f, group);
    struct mf_local* local = f->locals;
    while (local) { struct mf_local* next = local->next; if (local->group == group) mf_local_free(local); local = next; }
    for (;;) {
        struct mf_pending* found = NULL;
        for (struct mf_pending* p = f->pending; p; p = p->next) if (p->group == group) { found = p; break; }
        if (!found) break;
        member_files_cancel(found);
    }
    struct mf_job* j = f->jobs;
    while (j) { struct mf_job* next = j->next; if (j->group == group && j != f->worker) mf_job_free(j); j = next; }
}

static void mf_gui_done(void* arg, const struct mf_message* response) {
    struct mf_gui_request* q = arg;
    uint8_t event[16 + MF_WIRE_MAX]; mf_put64(event, q->group); mf_put64(event + 8, q->server);
    int n = mf_encode(response, event + 16, MF_WIRE_MAX);
    DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: GUI result id=%llu cmd=%s status=%s bytes=%d",
        (unsigned long long)response->request_id, mf_command_text(response->command), mf_status_text(response->status), n);
    if (n > 0) chat_event_post(q->inst, CHAT_EVT_MEMBER_FILES_RESPONSE, event, n + 16);
    u_free(q);
}

void member_files_gui_request(void* arg) {
    struct mf_gui_request* q = arg;
    if (member_files_request(q->inst, q->group, q->server, &q->message, mf_gui_done, q)) return;
    q->message.kind = MF_RESPONSE; q->message.status = MF_ACCESS_DENIED; q->message.data_len = 0;
    mf_gui_done(q, &q->message);
}