/* Личные сообщения проходят через попарные группы; общей группы A/S/M/B нет.
 * Подписанные участники, настоящий ETCP/router и отдельная SQLite каждого узла. */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "utun_instance.h"
#include "config_updater.h"
#include "chat/chat_core.h"
#include "chat/member_sync.h"
#include "chat/chat_setting.h"
#include "dm/dm_core.h"
#include "dm/dm_mailbox.h"
#include "dm/dm_media.h"
#include "routing_layer/topo_group.h"
#include "routing_layer/topo_node_sqlite.h"
#include "routing_layer/etcp_router.h"
#include "routing_layer/route_crypto.h"
#include "transport_layer/etcp.h"
#include "transport_layer/etcp_api.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include "test_net.h"
#include "test_membership.h"
#include "test_utils.h"

enum { A, S, M, B, NODES };
static const unsigned pairs[][2] = {{A,S}, {A,S}, {S,B}, {S,M}, {A,M}, {M,B}, {A,B}, {S,M}, {M,B}};
#define GROUPS (sizeof(pairs) / sizeof(pairs[0]))
#define REQUIRE(x) do { if (!(x)) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "DM routes: %s line=%d", #x, __LINE__); return -1; } } while (0)
#define WAIT(x) do { uint64_t end = get_time_tb() + 70000; while (!(x) && get_time_tb() < end) uasync_poll(f.ua, 1); REQUIRE(x); } while (0)

static struct {
    struct UASYNC* ua;
    struct UTUN_INSTANCE* node[NODES];
    struct TOPO_GROUP* group[NODES][GROUPS];
    char directory[512], conv[64];
    etcp_recv_fn receiver[NODES];
    unsigned media_seen[NODES];
    unsigned have_seen;
    unsigned observed, passed, failed;
} f;

static void check(const char* name, int ok) {
    if (ok) { f.passed++; DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS %s", name); }
    else { f.failed++; DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "FAIL %s", name); }
}

static int value(unsigned node, const char* sql) {
    sqlite3_stmt* st = NULL;
    int rc = sqlite3_prepare_v2(f.node[node]->topo_sqlite_db, sql, -1, &st, NULL), result = -1;
    if (rc == SQLITE_OK) rc = sqlite3_step(st);
    if (rc == SQLITE_ROW) result = sqlite3_column_int(st, 0);
    int final = sqlite3_finalize(st);
    if (rc != SQLITE_ROW || final != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "DM routes SQL: %s rc=%d final=%d", sql, rc, final);
    return result;
}

static int member(unsigned node, unsigned group, unsigned peer, const char* field, int state) {
    char sql[192];
    snprintf(sql, sizeof(sql), "UPDATE peers_%u SET %s=%d WHERE node_id=%lld", 101 + group, field, state,
        (long long)f.node[peer]->node_id);
    REQUIRE(sqlite3_exec(f.node[node]->topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK);
    return 0;
}

static void observed(struct ETCP_CONN* conn, struct ll_entry* entry) {
    for (unsigned i = 0; i < NODES; i++) if (conn->instance == f.node[i]) {
        if (entry->len >= sizeof(struct SVC_ROUTE_HDR)) {
            uint8_t svc = entry->dgram[offsetof(struct SVC_ROUTE_HDR, svc_id)];
            if (i == S && svc == ETCP_RT_ID_DM_MAILBOX) f.observed++;
            if (svc == ETCP_RT_ID_DM_MEDIA) f.media_seen[i]++;
            if (i == M && svc == ETCP_RT_ID_DM_MEDIA && entry->len > SVC_ROUTE_HDR_SIZE &&
                entry->dgram[SVC_ROUTE_HDR_SIZE] == DM_MEDIA_HAVE) f.have_seen++;
        }
        f.receiver[i](conn, entry); return;
    }
}

/* Инъекция подписана настоящим A; обходим только исходящую политику отправителя. */
static int inject(unsigned group, const uint8_t* payload, size_t len, int signed_packet) {
    struct SVC_ROUTE_HDR header = {.cmd = ETCP_ID_SVC_ROUTE, .group_id = 101 + group,
        .src_node_id = f.node[A]->node_id, .dst_node_id = f.node[S]->node_id,
        .svc_id = ETCP_RT_ID_DM_MAILBOX, .reset_id = 123, .flags = ROUTER_FLAG_START};
    uint8_t plain[2048]; REQUIRE(sizeof(header) + len <= sizeof(plain));
    memcpy(plain, &header, sizeof(header)); memcpy(plain + sizeof(header), payload, len);
    uint8_t* encoded = NULL; size_t size = 0;
    REQUIRE(!route_crypto_encode(f.node[A], header.dst_node_id, plain, sizeof(header) + len,
        signed_packet ? ROUTE_CRYPTO_SIGN : 0, &encoded, &size));
    struct ll_entry* entry = queue_entry_new(0); REQUIRE(entry);
    entry->dgram = u_malloc(size + 9); REQUIRE(entry->dgram);
    memcpy(entry->dgram, encoded, size); u_free(encoded);
    memcpy(entry->dgram + size, &header.src_node_id, 8); entry->dgram[size + 8] = 1;
    entry->len = (uint16_t)(size + 9);
    struct ETCP_CONN* conn = instance_find_conn(f.node[A], f.node[S]->node_id); REQUIRE(conn && conn->links_up);
    unsigned before = f.observed;
    REQUIRE(!etcp_send(conn, entry));
    WAIT(f.observed > before);
    return 0;
}

static int setup(void) {
    snprintf(f.directory, sizeof(f.directory), "/tmp/utun_dm_routes_XXXXXX"); REQUIRE(!test_mkdtemp(f.directory));
    struct SC_MYKEYS keys[NODES]; char pub[NODES][65], priv[NODES][65];
    for (unsigned i = 0; i < NODES; i++) {
        REQUIRE(!sc_generate_keypair(&keys[i]));
        bytes_to_hex(keys[i].public_key, 32, pub[i], 65); bytes_to_hex(keys[i].private_key, 32, priv[i], 65);
    }
    f.ua = uasync_create(); REQUIRE(f.ua);
    for (unsigned i = 0; i < NODES; i++) {
        char config[4096];
        int len = snprintf(config, sizeof(config), "[global]\nauto_sockets=no\nmy_public_key=%s\nmy_private_key=%s\n"
            "db_path=%s/node%u\n[server: udp]\naddr=127.0.0.1:0\ntype=public\n[allowed_keys]\nallow_all=1\n"
            "[chatserver]\ngroup_autoconnect=0\nstorage_autoload=1\n", pub[i], priv[i], f.directory, i);
        for (unsigned j = 0; j < i; j++) len += snprintf(config + len, sizeof(config) - len,
            "[client: peer%u]\npeer_public_key=%s\nlink=udp:127.0.0.1:0\n", j, pub[j]);
        f.node[i] = utun_instance_create_from_str(f.ua, config); REQUIRE(f.node[i]);
    }
    for (unsigned i = 0; i < NODES; i++) {
        for (unsigned j = 0; j < i; j++) {
            char name[32]; snprintf(name, sizeof(name), "peer%u", j); test_etcp_set_peer(f.node[i], name, f.node[j], "udp");
        }
        REQUIRE(!utun_core_start(f.node[i]) && !chat_service_start(f.node[i]));
    }
    struct UTUN_INSTANCE* owner = f.node[A];
    for (unsigned g = 0; g < GROUPS; g++) {
        char channel[32]; snprintf(channel, sizeof(channel), "%u", 101 + g);
        uint8_t signature[64];
        REQUIRE(!sc_ed25519_sign(owner->my_ed25519_privkey, (const uint8_t*)channel, strlen(channel), signature));
        for (unsigned side = 0; side < 2; side++) {
            unsigned local = pairs[g][side]; struct UTUN_INSTANCE* inst = f.node[local];
            REQUIRE(!topo_node_sqlite_channel_put(inst->topo_sqlite_db, channel, "dm-routes", owner->node_id,
                owner->my_keys.public_key, NULL, owner->my_ed25519_pubkey, NULL, signature));
            for (unsigned p = 0; p < 2; p++) {
                unsigned index = pairs[g][p]; struct UTUN_INSTANCE* node = f.node[index];
                const char* tags = index == S && g != 1 ? "{\"supernode\":\"yes\",\"ver\":\"1\"}" :
                    index == M ? "{\"storage\":\"yes\",\"ver\":\"1\"}" : "{\"ver\":\"1\"}";
                uint8_t body[256], tag_signature[64];
                int size = member_sync_build_owner_msg(node->node_id, tags, body, sizeof(body)); REQUIRE(size > 0);
                REQUIRE(!sc_ed25519_sign(owner->my_ed25519_privkey, body, size, tag_signature));
                REQUIRE(test_member_seed(inst, channel, node->my_keys.public_key, node->my_ed25519_pubkey,
                    node->my_ed25519_privkey, owner->my_ed25519_privkey, "{}", tags, tag_signature) >= 0);
            }
            REQUIRE(!chat_core_ensure_channel_ready(inst, channel));
            f.group[local][g] = topo_groups_find(inst->topo_groups, 101 + g);
            REQUIRE(f.group[local][g]);
        }
    }
    for (unsigned g = 0; g < GROUPS; g++) for (unsigned side = 0; side < 2; side++) {
        unsigned local = pairs[g][side], peer = pairs[g][1 - side];
        struct ETCP_CONN* conn;
        WAIT((conn = instance_find_conn(f.node[local], f.node[peer]->node_id)) && conn->links_up);
        REQUIRE(!topo_group_new_conn(f.group[local][g], conn));
    }
    for (unsigned g = 0; g < GROUPS; g++) for (unsigned side = 0; side < 2; side++)
        WAIT(topo_group_peer_ready(f.group[pairs[g][side]][g], f.node[pairs[g][1-side]]->node_id));
    for (unsigned i = 0; i < NODES; i++) {
        f.receiver[i] = f.node[i]->api_bindings.callbacks[ETCP_ID_SVC_ROUTE]; REQUIRE(f.receiver[i]);
        f.node[i]->api_bindings.callbacks[ETCP_ID_SVC_ROUTE] = observed;
    }
    uint64_t conv;
    REQUIRE(!dm_start(owner, f.node[B]->node_id, f.node[B]->my_keys.public_key, f.node[B]->my_ed25519_pubkey,
        "recipient", "107", DM_MODE_DEVICE, &conv));
    snprintf(f.conv, sizeof(f.conv), "%llu", (unsigned long long)conv);
    return 0;
}

static int run(void) {
    uint64_t route = 0;
    check("role in 101 permits S while route uses 102", dm_peer_role(f.node[A], f.node[S]->node_id, "supernode") &&
        dm_route_group(f.node[A], f.node[S]->node_id, 102, &route) && route == 102);
    const char* fields[] = {"admitted", "deleted", "source"};
    for (unsigned i = 0; i < 3; i++) {
        REQUIRE(!member(A, 0, S, fields[i], i ? 1 : 0));
        check("unadmitted/deleted/foreign-source row gives no role", !dm_peer_role(f.node[A], f.node[S]->node_id, "supernode"));
        REQUIRE(!member(A, 0, S, fields[i], i ? 0 : 1));
    }
    REQUIRE(!member(A, 0, A, "admitted", 0));
    check("self revocation removes role from that group", !dm_peer_role(f.node[A], f.node[S]->node_id, "supernode"));
    REQUIRE(!member(A, 0, A, "admitted", 1));
    /* Отдельная A/B-группа нужна только для начала беседы, затем прямого маршрута нет. */
    REQUIRE(!member(A, 6, B, "admitted", 0) && !member(B, 6, A, "admitted", 0));
    /* Синхронизация подписанного состава не должна восстановить прямой путь в этом сценарии. */
    f.group[A][6]->stopping = f.group[B][6]->stopping = 1;
    topo_group_revalidate_members(f.group[A][6]); topo_group_revalidate_members(f.group[B][6]);
    REQUIRE(!dm_send(f.node[A], f.conv, "text", (const uint8_t*)"pairwise mailbox", 16));
    uint8_t packet[2048] = {1};
    memcpy(packet + 1, &f.node[B]->node_id, 8);
    sqlite3_stmt* st = NULL;
    REQUIRE(sqlite3_prepare_v2(f.node[A]->topo_sqlite_db, "SELECT body FROM dm_outbox", -1, &st, NULL) == SQLITE_OK);
    REQUIRE(sqlite3_step(st) == SQLITE_ROW); int bytes = sqlite3_column_bytes(st, 0);
    REQUIRE(bytes > 0 && bytes + 9 <= (int)sizeof(packet)); memcpy(packet + 9, sqlite3_column_blob(st, 0), bytes);
    REQUIRE(sqlite3_finalize(st) == SQLITE_OK);
    REQUIRE(!member(S, 1, A, "admitted", 0));
    REQUIRE(!inject(1, packet, bytes + 9, 1));
    check("signed PUT through revoked group does not store pending", value(S, "SELECT count(*) FROM dm_pending_messages") == 0);
    REQUIRE(!member(S, 1, A, "admitted", 1));
    REQUIRE(!inject(1, packet, bytes + 9, 0));
    check("unsigned mailbox PUT does not store pending", value(S, "SELECT count(*) FROM dm_pending_messages") == 0);
    struct ETCP_ROUTER_CONN* conn = etcp_router_conn_get(f.node[A], 102, f.node[S]->node_id, ETCP_RT_ID_DM_MAILBOX); REQUIRE(conn);
    REQUIRE(!etcp_router_conn_send(conn, packet, bytes + 9, ROUTER_FLAG_SIGNED));
    WAIT(value(B, "SELECT count(*) FROM dm_messages") == 1);
    WAIT(value(A, "SELECT count(*) FROM dm_outbox") == 0);
    check("mailbox crosses A/S and S/B groups without common A/S/B group",
        value(S, "SELECT count(*) FROM dm_pending_messages") == 0 && value(S, "SELECT count(*) FROM dm_mail_receipts") == 1);
    conn = etcp_router_conn_get(f.node[A], 101, f.node[S]->node_id, ETCP_RT_ID_DM_MAILBOX); REQUIRE(conn);
    REQUIRE(!etcp_router_conn_send(conn, packet, bytes + 9, ROUTER_FLAG_SIGNED));
    uint64_t end = get_time_tb() + 2000; while (get_time_tb() < end) uasync_poll(f.ua, 1);
    check("same PUT over another group preserves receipt and single message", value(S, "SELECT count(*) FROM dm_pending_messages") == 0 &&
        value(B, "SELECT count(*) FROM dm_messages") == 1 && value(S, "SELECT count(*) FROM dm_mail_receipts") == 1);
    check("mailbox persists identities without route column", value(S,
        "SELECT count(*) FROM pragma_table_info('dm_pending_messages') WHERE name='group_id'") == 0);
    return 0;
}

static int file_delivery(void) {
    REQUIRE(!chat_setting_set(f.node[B], "storage_autoload", "0"));
    char path[768]; snprintf(path, sizeof(path), "%s/source.bin", f.directory);
    FILE* source = fopen(path, "wb"); REQUIRE(source);
    for (unsigned i = 0; i < 131079; i++) REQUIRE(fputc((i * 73 + 19) & 255, source) != EOF);
    REQUIRE(!fclose(source));
    struct attachment_info info = {.kind = ATTACHMENT_FILE};
    snprintf(info.name, sizeof(info.name), "pairwise.bin");
    REQUIRE(!dm_send_file(f.node[A], f.conv, path, &info, 0, 0));
    WAIT(value(A, "SELECT count(*) FROM dm_files WHERE ready=1") == 1);
    sqlite3_stmt* st = NULL;
    REQUIRE(sqlite3_prepare_v2(f.node[A]->topo_sqlite_db, "SELECT body FROM dm_files WHERE ready=1", -1, &st, NULL) == SQLITE_OK);
    REQUIRE(sqlite3_step(st) == SQLITE_ROW);
    int bytes = sqlite3_column_bytes(st, 0); uint8_t body[8192];
    REQUIRE(bytes > 0 && bytes <= (int)sizeof(body)); memcpy(body, sqlite3_column_blob(st, 0), bytes);
    REQUIRE(sqlite3_finalize(st) == SQLITE_OK);
    REQUIRE(!dm_mailbox_store_file(f.node[A], f.node[B]->node_id, body, bytes));
    WAIT(value(M, "SELECT count(*) FROM dm_custody WHERE ready=1") == 1 && value(B, "SELECT count(*) FROM dm_files") == 1);
    REQUIRE(value(B, "SELECT ready FROM dm_files") == 0);
    /* HAVE уже отправлен, но окно выбора источника ещё открыто. Отзыв отменяет эту пробу. */
    WAIT(value(B, "SELECT count(*) FROM dm_file_sources WHERE expires>0") == 1);
    uint64_t seq; memcpy(&seq, body + 8, 8);
    unsigned seen = f.have_seen;
    REQUIRE(!dm_media_user_action(f.node[B], f.conv, f.node[A]->node_id, seq, 0));
    WAIT(f.have_seen > seen);
    struct ETCP_CONN* transport = instance_find_conn(f.node[B], f.node[M]->node_id); REQUIRE(transport && transport->links_up);
    f.group[B][5]->stopping = f.group[M][5]->stopping = 1;
    topo_group_revalidate_members(f.group[B][5]); topo_group_revalidate_members(f.group[M][5]);
    WAIT(value(B, "SELECT count(*) FROM dm_files WHERE ready=1") == 1);
    check("revoked HAVE probe retries another group while ETCP and custody survive",
        instance_find_conn(f.node[B], f.node[M]->node_id) == transport && transport->links_up &&
        value(M, "SELECT count(*) FROM dm_custody WHERE ready=1") == 1);
    REQUIRE(sqlite3_prepare_v2(f.node[B]->topo_sqlite_db, "SELECT local_path FROM dm_files WHERE ready=1", -1, &st, NULL) == SQLITE_OK);
    REQUIRE(sqlite3_step(st) == SQLITE_ROW); snprintf(path, sizeof(path), "%s", sqlite3_column_text(st, 0));
    REQUIRE(sqlite3_finalize(st) == SQLITE_OK);
    FILE* received = fopen(path, "rb"); REQUIRE(received);
    int exact = 1;
    for (unsigned i = 0; i < 131079; i++) if (fgetc(received) != (int)((i * 73 + 19) & 255)) { exact = 0; break; }
    if (exact && fgetc(received) != EOF) exact = 0;
    REQUIRE(!fclose(received));
    check("encrypted file crosses A/S, S/M, A/M, S/B and M/B groups byte for byte", exact);
    check("copy is stored once on M and one assignment exists on S", value(M, "SELECT count(*) FROM dm_custody WHERE ready=1") == 1 &&
        value(S, "SELECT count(*) FROM dm_media_holders") == 1);
    return 0;
}

static int storage_assignments(void) {
    uint8_t request[24 + 8192]; sqlite3_stmt* st = NULL;
    REQUIRE(sqlite3_prepare_v2(f.node[S]->topo_sqlite_db,
        "SELECT h.assignment_id,j.body FROM dm_media_holders h JOIN dm_media_jobs j USING(id)", -1, &st, NULL) == SQLITE_OK);
    REQUIRE(sqlite3_step(st) == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 16);
    memcpy(request, sqlite3_column_blob(st, 0), 16); memcpy(request + 16, &f.node[B]->node_id, 8);
    int len = sqlite3_column_bytes(st, 1); REQUIRE(len > 0 && len <= 8192);
    memcpy(request + 24, sqlite3_column_blob(st, 1), len); REQUIRE(sqlite3_finalize(st) == SQLITE_OK);
    int expires = value(M, "SELECT expires FROM dm_custody"), policy = value(M, "SELECT delete_on_open FROM dm_custody");
    REQUIRE(expires > 0 && policy >= 0);
    REQUIRE(!chat_setting_set(f.node[M], "dm_media_ttl_sec", "600") &&
        !chat_setting_set(f.node[M], "dm_media_delete_on_open", policy ? "no" : "yes"));
    /* После отзыва первого пути прежнее назначение продолжается через 108 при том же ETCP. */
    struct ETCP_CONN* transport = instance_find_conn(f.node[S], f.node[M]->node_id); REQUIRE(transport && transport->links_up);
    f.group[S][3]->stopping = f.group[M][3]->stopping = 1;
    topo_group_revalidate_members(f.group[S][3]); topo_group_revalidate_members(f.group[M][3]);
    uint64_t route = 0; REQUIRE(dm_route_group(f.node[S], f.node[M]->node_id, 104, &route) && route == 108);
    unsigned seen = f.media_seen[S];
    REQUIRE(!dm_media_send(f.node[S], 104, f.node[M]->node_id, DM_MEDIA_STORE, request, len + 24));
    WAIT(f.media_seen[S] > seen);
    check("revoked S/M group keeps assignment and physical ETCP through second group",
        instance_find_conn(f.node[S], f.node[M]->node_id) == transport && transport->links_up &&
        value(M, "SELECT count(*) FROM dm_custody_requests") == 1 && value(M, "SELECT expires FROM dm_custody") == expires);
    f.group[S][7]->stopping = f.group[M][7]->stopping = 1;
    topo_group_revalidate_members(f.group[S][7]); topo_group_revalidate_members(f.group[M][7]);
    check("no pair route pauses STORE without changing reservation", dm_media_send(f.node[S], 0, f.node[M]->node_id,
        DM_MEDIA_STORE, request, len + 24) != 0 && value(M, "SELECT count(*) FROM dm_custody_requests") == 1 &&
        value(M, "SELECT expires FROM dm_custody") == expires);
    f.group[S][7]->stopping = f.group[M][7]->stopping = 0;
    topo_group_revalidate_members(f.group[S][7]); topo_group_revalidate_members(f.group[M][7]);
    /* A дополнительно становится вторым S только в своей группе с M. */
    const char* tags = "{\"supernode\":\"yes\",\"ver\":\"2\"}";
    uint8_t owner_body[256], signature[64];
    int size = member_sync_build_owner_msg(f.node[A]->node_id, tags, owner_body, sizeof(owner_body)); REQUIRE(size > 0);
    REQUIRE(!sc_ed25519_sign(f.node[A]->my_ed25519_privkey, owner_body, size, signature));
    for (unsigned i = 0; i < 2; i++) REQUIRE(test_member_seed(f.node[i ? M : A], "105", f.node[A]->my_keys.public_key,
        f.node[A]->my_ed25519_pubkey, f.node[A]->my_ed25519_privkey, f.node[A]->my_ed25519_privkey, "{}", tags, signature) >= 0);
    REQUIRE(dm_peer_role(f.node[M], f.node[A]->node_id, "supernode"));
    memset(request, 0x55, 16); seen = f.media_seen[A];
    REQUIRE(!dm_media_send(f.node[A], 105, f.node[M]->node_id, DM_MEDIA_STORE, request, len + 24));
    WAIT(value(M, "SELECT count(*) FROM dm_custody_requests") == 2 && f.media_seen[A] > seen);
    check("two coordinators share one copy with first deadline and policy",
        value(M, "SELECT count(*) FROM dm_custody WHERE ready=1") == 1 &&
        value(M, "SELECT expires FROM dm_custody") == expires && value(M, "SELECT delete_on_open FROM dm_custody") == policy);
    REQUIRE(!member(M, 4, A, "admitted", 0));
    /* Без polling: исходящий путь A ещё допустим, входящий M проверяет свежую строку. */
    seen = f.media_seen[M]; request[0]++;
    REQUIRE(!dm_media_send(f.node[A], 105, f.node[M]->node_id, DM_MEDIA_STORE, request, len + 24));
    WAIT(f.media_seen[M] > seen);
    check("revoked requester cannot create another assignment", value(M, "SELECT count(*) FROM dm_custody_requests") == 2);
    REQUIRE(!member(M, 4, A, "admitted", 1));
    /* Новый объект проверяет реальный цикл S: durable отказ, пауза, новый ID на том же M. */
    f.group[S][2]->stopping = f.group[B][2]->stopping = 1;
    topo_group_revalidate_members(f.group[S][2]); topo_group_revalidate_members(f.group[B][2]);
    uint8_t* body = request + 24;
    uint8_t* desc = (uint8_t*)dm_message_media(body, len); REQUIRE(desc);
    desc[0] ^= 0x80; uint64_t plain = 2 * 1024 * 1024; memcpy(desc + 16, &plain, 8);
    uint64_t seq; memcpy(&seq, body + 8, 8); seq++; memcpy(body + 8, &seq, 8);
    REQUIRE(!sc_ed25519_sign(f.node[A]->my_ed25519_privkey, body, len - DM_MSG_SIG_SIZE, body + len - DM_MSG_SIG_SIZE));
    REQUIRE(!chat_setting_set(f.node[M], "dm_media_storage_mb", "1"));
    REQUIRE(!dm_mailbox_store_file(f.node[A], f.node[B]->node_id, body, len));
    WAIT(value(S, "SELECT count(*) FROM dm_media_holders WHERE released=3") == 1);
    REQUIRE(sqlite3_prepare_v2(f.node[S]->topo_sqlite_db,
        "SELECT assignment_id,id FROM dm_media_holders WHERE released=3", -1, &st, NULL) == SQLITE_OK);
    REQUIRE(sqlite3_step(st) == SQLITE_ROW);
    uint8_t old_reply[33]; memcpy(old_reply, sqlite3_column_blob(st, 0), 16); memcpy(old_reply + 16, sqlite3_column_blob(st, 1), 16);
    old_reply[32] = 1; REQUIRE(sqlite3_finalize(st) == SQLITE_OK);
    check("quota refusal is durable and schedules delayed retry", value(M, "SELECT count(*) FROM dm_custody_requests WHERE rejected=1") == 1 &&
        value(S, "SELECT count(*) FROM dm_media_jobs WHERE holder=0 AND retry_at>0") == 1);
    REQUIRE(!chat_setting_set(f.node[M], "dm_media_storage_mb", "1024"));
    REQUIRE(sqlite3_exec(f.node[S]->topo_sqlite_db, "UPDATE dm_media_jobs SET retry_at=0 WHERE holder=0", NULL, NULL, NULL) == SQLITE_OK);
    WAIT(value(M, "SELECT count(*) FROM dm_custody") == 2 && value(S, "SELECT count(*) FROM dm_media_jobs WHERE delete_on_open>=0") == 2);
    REQUIRE(sqlite3_prepare_v2(f.node[S]->topo_sqlite_db,
        "SELECT assignment_id FROM dm_media_holders WHERE id=?", -1, &st, NULL) == SQLITE_OK);
    REQUIRE(sqlite3_bind_blob(st, 1, old_reply + 16, 16, SQLITE_STATIC) == SQLITE_OK && sqlite3_step(st) == SQLITE_ROW);
    int different = sqlite3_column_bytes(st, 0) == 16 && memcmp(old_reply, sqlite3_column_blob(st, 0), 16);
    REQUIRE(sqlite3_finalize(st) == SQLITE_OK);
    check("new attempt on same storage has new ID while old refusal remains", different &&
        value(M, "SELECT count(*) FROM dm_custody_requests") == 4 && value(M, "SELECT count(*) FROM dm_custody_requests WHERE rejected=1") == 1);
    seen = f.media_seen[S];
    REQUIRE(!dm_media_send(f.node[M], 108, f.node[S]->node_id, DM_MEDIA_REJECT, old_reply, sizeof(old_reply)));
    WAIT(f.media_seen[S] > seen);
    check("old rejection cannot reset new reservation on same storage",
        value(S, "SELECT count(*) FROM dm_media_jobs WHERE holder<>0 AND delete_on_open>=0") == 2 &&
        value(S, "SELECT count(*) FROM dm_media_holders WHERE released<>0") == 0);
    return 0;
}

int main(void) {
    debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO);
    if (getenv("UTUN_TEST_DEBUG")) {
        debug_set_category_level(DEBUG_CATEGORY_DM, DEBUG_LEVEL_TRACE);
        debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_DEBUG);
        debug_set_category_level(DEBUG_CATEGORY_BGP, DEBUG_LEVEL_INFO);
    }
    size_t before = u_get_allocated_count();
    if (setup() || run() || file_delivery() || storage_assignments()) f.failed++;
    for (unsigned i = 0; i < NODES; i++) if (f.node[i]) utun_instance_destroy(f.node[i]);
    if (f.ua) uasync_destroy(f.ua, 0);
    check("fixture releases all allocations", before == u_get_allocated_count());
    printf("test_dm_group_routes: %u PASS %u FAIL artifacts=%s\n", f.passed, f.failed, f.directory);
    return f.failed ? 1 : 0;
}