/* Настоящие core/SQLite/подписи; вход после router, независимый reader в ACK/событии. */
#define _GNU_SOURCE
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <openssl/sha.h>
#include "../src/utun_instance.h"
#include "../src/chat/chat_core.h"
#include "../src/chat/chat_attachment.h"
#include "../src/chat/chat_core_priv.h"
#include "../src/chat/chat_event.h"
#include "../src/chat/member_sync.h"
#include "../src/routing_layer/topo_node_sqlite.h"
#include "../src/routing_layer/etcp_router.h"
#include "../src/transport_layer/etcp.h"
#include "../lib/platform_compat.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"

enum operation { NO_OPERATION, IDENTITY, MEMBER, AUTHOR, READ, INSERT, OUTBOX, PENDING, ELIGIBLE, PUMP_READ, PUMP_RETRY };
enum failure { NO_FAILURE, BEGIN, PREPARE, BIND, STEP, FINALIZE, CHANGES, COMMIT };
static enum failure fault;
static enum operation fault_operation, last_operation;
static int fault_index, fail_rollback;
static unsigned fault_hits, cases;

struct snapshot {
    uint8_t key[32];
    int exists, incoming, state, command, error, key_bytes;
    int64_t created, retry;
};
struct fixture {
    struct UASYNC* ua;
    struct UTUN_INSTANCE* inst;
    struct ETCP_ROUTER_CONN route; /* Только наблюдение ACK; сеть/crypto проверяет encrypted_attachment. */
    sqlite3* reader;
    char directory[256], channel[32];
    uint8_t message[32], id[16], hash[32], key[32];
    uint64_t peer;
    struct snapshot expected;
    int64_t started;
    int allow_result, route_up, room, send_rc, corrupt, flags, outgoing, pump_only;
    unsigned acks, events;
};
static struct fixture* active;
static timeout_callback_t attachment_tick;

/* Вызываем зарегистрированный production callback без ожидания секунды; старый таймер отменяем. */
extern void* __real_uasync_set_timeout(struct UASYNC*, int, void*, timeout_callback_t, const char*);
void* __wrap_uasync_set_timeout(struct UASYNC* ua, int delay, void* arg, timeout_callback_t callback, const char* name) {
    if (name && !strcmp(name, "attachment keys")) attachment_tick = callback;
    return __real_uasync_set_timeout(ua, delay, arg, callback, name);
}

static enum operation operation(const char* query) {
    if (!query) return NO_OPERATION;
    static const char* prefixes[] = {NULL, "SELECT crypto_id,cipher_hash,key FROM attachment_keys WHERE",
        "SELECT x25519_pubkey,ed25519_pubkey,signed_by,signature,join_sig,join_ts", "SELECT 1 FROM msg_",
        "SELECT incoming,state FROM attachment_key_requests WHERE", "INSERT INTO attachment_key_requests(",
        "INSERT INTO attachment_key_outbox(", "SELECT channel,message_id,crypto_id,author,key,cipher_hash,cipher_path,plain_path",
        "SELECT 1 FROM attachment_keys a WHERE", "SELECT channel,message_id,peer,command FROM attachment_key_outbox",
        "UPDATE attachment_key_outbox SET retry_at=unixepoch()+5 WHERE"};
    for (enum operation op = IDENTITY; op <= PUMP_RETRY; op++)
        if (!strncmp(query, prefixes[op], strlen(prefixes[op]))) return op;
    return NO_OPERATION;
}

static int take(enum failure point, enum operation op, int index) {
    if (fault != point || fault_operation != op || (point == BIND && fault_index != index)) return 0;
    fault = NO_FAILURE; fault_hits++;
    DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "request test: injected point=%d operation=%d bind=%d", point, op, index); return 1;
}

extern int __real_sqlite3_exec(sqlite3*, const char*, int (*)(void*, int, char**, char**), void*, char**);
int __wrap_sqlite3_exec(sqlite3* db, const char* query, int (*cb)(void*, int, char**, char**), void* arg, char** error) {
    if (active && db == active->inst->topo_sqlite_db) {
        const char* begin = active->pump_only ? "BEGIN IMMEDIATE /* attachment outbox */" : "BEGIN IMMEDIATE";
        const char* commit = active->pump_only ? "COMMIT /* attachment outbox */" : "COMMIT";
        if (!strcmp(query, begin) && take(BEGIN, NO_OPERATION, 0)) return SQLITE_BUSY;
        if (!strcmp(query, commit) && take(COMMIT, NO_OPERATION, 0)) return SQLITE_BUSY;
        if (!strcmp(query, "ROLLBACK") && fail_rollback) {
            fail_rollback = 0; DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "request test: injected rollback failure"); return SQLITE_IOERR;
        }
    }
    return __real_sqlite3_exec(db, query, cb, arg, error);
}

extern int __real_sqlite3_prepare_v2(sqlite3*, const char*, int, sqlite3_stmt**, const char**);
int __wrap_sqlite3_prepare_v2(sqlite3* db, const char* query, int length, sqlite3_stmt** st, const char** tail) {
    enum operation op = operation(query);
    if (op && take(PREPARE, op, 0)) { *st = NULL; return SQLITE_NOMEM; }
    return __real_sqlite3_prepare_v2(db, query, length, st, tail);
}

extern int __real_sqlite3_bind_blob(sqlite3_stmt*, int, const void*, int, void (*)(void*));
int __wrap_sqlite3_bind_blob(sqlite3_stmt* st, int index, const void* value, int length, void (*destroy)(void*)) {
    if (take(BIND, operation(sqlite3_sql(st)), index)) return SQLITE_NOMEM;
    return __real_sqlite3_bind_blob(st, index, value, length, destroy);
}

extern int __real_sqlite3_bind_text(sqlite3_stmt*, int, const char*, int, void (*)(void*));
int __wrap_sqlite3_bind_text(sqlite3_stmt* st, int index, const char* value, int length, void (*destroy)(void*)) {
    if (take(BIND, operation(sqlite3_sql(st)), index)) return SQLITE_NOMEM;
    return __real_sqlite3_bind_text(st, index, value, length, destroy);
}

extern int __real_sqlite3_bind_int64(sqlite3_stmt*, int, sqlite3_int64);
int __wrap_sqlite3_bind_int64(sqlite3_stmt* st, int index, sqlite3_int64 value) {
    if (take(BIND, operation(sqlite3_sql(st)), index)) return SQLITE_NOMEM;
    return __real_sqlite3_bind_int64(st, index, value);
}

extern int __real_sqlite3_bind_int(sqlite3_stmt*, int, int);
int __wrap_sqlite3_bind_int(sqlite3_stmt* st, int index, int value) {
    if (take(BIND, operation(sqlite3_sql(st)), index)) return SQLITE_NOMEM;
    return __real_sqlite3_bind_int(st, index, value);
}

extern int __real_sqlite3_step(sqlite3_stmt*);
int __wrap_sqlite3_step(sqlite3_stmt* st) {
    last_operation = operation(sqlite3_sql(st));
    if (last_operation && take(STEP, last_operation, 0)) return SQLITE_IOERR;
    return __real_sqlite3_step(st);
}

extern int __real_sqlite3_finalize(sqlite3_stmt*);
int __wrap_sqlite3_finalize(sqlite3_stmt* st) {
    enum operation op = st ? operation(sqlite3_sql(st)) : NO_OPERATION;
    int failed = op && take(FINALIZE, op, 0);
    int rc = __real_sqlite3_finalize(st); return failed ? SQLITE_IOERR : rc;
}

extern int __real_sqlite3_changes(sqlite3*);
int __wrap_sqlite3_changes(sqlite3* db) {
    if (last_operation && take(CHANGES, last_operation, 0)) return 0;
    return __real_sqlite3_changes(db);
}

static void sql(struct fixture* f, const char* query) {
    int rc = sqlite3_exec(f->inst->topo_sqlite_db, query, NULL, NULL, NULL);
    if (rc != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "request test SQL rc=%d: %s", rc, sqlite3_errmsg(f->inst->topo_sqlite_db));
    assert(rc == SQLITE_OK);
}

/* Проверяем SQLite timestamps её же часами: libc time() может отставать у границы секунды. */
static int64_t database_time(struct fixture* f) {
    sqlite3_stmt* st = NULL;
    assert(sqlite3_prepare_v2(f->reader, "SELECT unixepoch()", -1, &st, NULL) == SQLITE_OK && sqlite3_step(st) == SQLITE_ROW);
    int64_t now = sqlite3_column_int64(st, 0);
    assert(sqlite3_step(st) == SQLITE_DONE && sqlite3_finalize(st) == SQLITE_OK); return now;
}

static struct snapshot snapshot(struct fixture* f) {
    sqlite3_stmt* st = NULL; struct snapshot result = {0};
    assert(sqlite3_prepare_v2(f->reader, "SELECT a.key,a.error,r.message_id IS NOT NULL,coalesce(r.incoming,-1),coalesce(r.state,-1),"
        "coalesce(r.created_at,0),coalesce(o.command,0),coalesce(o.retry_at,0) FROM attachment_keys a"
        " LEFT JOIN attachment_key_requests r ON r.channel=a.channel AND r.message_id=a.message_id AND r.peer=?3"
        " LEFT JOIN attachment_key_outbox o ON o.channel=a.channel AND o.message_id=a.message_id AND o.peer=?3"
        " WHERE a.channel=?1 AND a.message_id=?2", -1, &st, NULL) == SQLITE_OK);
    assert(sqlite3_bind_text(st, 1, f->channel, -1, SQLITE_STATIC) == SQLITE_OK);
    assert(sqlite3_bind_blob(st, 2, f->message, 32, SQLITE_STATIC) == SQLITE_OK);
    assert(sqlite3_bind_int64(st, 3, (sqlite3_int64)f->peer) == SQLITE_OK && sqlite3_step(st) == SQLITE_ROW);
    result.key_bytes = sqlite3_column_bytes(st, 0); assert(result.key_bytes == 0 || result.key_bytes == 32);
    if (result.key_bytes) memcpy(result.key, sqlite3_column_blob(st, 0), 32);
    result.error = sqlite3_column_int(st, 1); result.exists = sqlite3_column_int(st, 2);
    result.incoming = sqlite3_column_int(st, 3); result.state = sqlite3_column_int(st, 4);
    result.created = sqlite3_column_int64(st, 5); result.command = sqlite3_column_int(st, 6); result.retry = sqlite3_column_int64(st, 7);
    assert(sqlite3_step(st) == SQLITE_DONE && sqlite3_finalize(st) == SQLITE_OK);
    return result;
}

static void equal(struct fixture* f) {
    struct snapshot current = snapshot(f);
    if (f->expected.created < 0) {
        int64_t now = database_time(f);
        if (current.created < f->started || current.created > now)
            DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "request test: created=%lld outside SQLite bounds=%lld..%lld",
                (long long)current.created, (long long)f->started, (long long)now);
        assert(current.created >= f->started && current.created <= now); f->expected.created = current.created;
    }
    assert(current.key_bytes == f->expected.key_bytes && !memcmp(current.key, f->expected.key, 32));
    assert(current.error == f->expected.error && current.exists == f->expected.exists);
    assert(current.incoming == f->expected.incoming && current.state == f->expected.state && current.created == f->expected.created);
    assert(current.command == f->expected.command && current.retry == f->expected.retry);
    sqlite3_stmt* st = NULL;
    assert(sqlite3_prepare_v2(f->reader, "SELECT key,error FROM attachment_keys WHERE channel='900'", -1, &st, NULL) == SQLITE_OK);
    assert(sqlite3_step(st) == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 32 && sqlite3_column_int(st, 1) == 7);
    uint8_t key[32] = {99}; assert(!memcmp(sqlite3_column_blob(st, 0), key, 32));
    assert(sqlite3_step(st) == SQLITE_DONE && sqlite3_finalize(st) == SQLITE_OK);
}

extern int __real_etcp_router_has_route(struct UTUN_INSTANCE*, uint64_t, uint64_t);
int __wrap_etcp_router_has_route(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer) {
    if (active && inst == active->inst && group == active->route.group_id && peer == active->peer) return active->route_up;
    return __real_etcp_router_has_route(inst, group, peer);
}

extern struct ETCP_ROUTER_CONN* __real_etcp_router_conn_get(struct UTUN_INSTANCE*, uint64_t, uint64_t, uint8_t);
struct ETCP_ROUTER_CONN* __wrap_etcp_router_conn_get(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, uint8_t service) {
    if (active && inst == active->inst && group == active->route.group_id && peer == active->peer && service == ETCP_RT_ID_ATTACHMENT_KEYS)
        return &active->route;
    return __real_etcp_router_conn_get(inst, group, peer, service);
}

extern int __real_etcp_router_send_q_has_room(struct UTUN_INSTANCE*, uint64_t, uint64_t, uint8_t);
int __wrap_etcp_router_send_q_has_room(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, uint8_t service) {
    if (active && inst == active->inst && group == active->route.group_id && peer == active->peer && service == ETCP_RT_ID_ATTACHMENT_KEYS)
        return active->room;
    return __real_etcp_router_send_q_has_room(inst, group, peer, service);
}

extern int __real_etcp_router_conn_send(struct ETCP_ROUTER_CONN*, const uint8_t*, size_t, int);
int __wrap_etcp_router_conn_send(struct ETCP_ROUTER_CONN* conn, const uint8_t* data, size_t length, int flags) {
    if (!active || conn != &active->route) return __real_etcp_router_conn_send(conn, data, length, flags);
    assert(active->allow_result && length == 85 && !memcmp(data, "AK01", 4) && data[4] == (active->outgoing ? 1 : 2));
    assert(flags == (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED) && sqlite3_get_autocommit(active->inst->topo_sqlite_db));
    assert(!memcmp(data + 5, active->message, 32) && !memcmp(data + 37, active->id, 16) && !memcmp(data + 53, active->hash, 32));
    if (active->outgoing) {
        active->expected.retry = snapshot(active).retry;
        assert(active->expected.retry >= active->started + 5 && active->expected.retry <= database_time(active) + 5);
    }
    equal(active); active->acks++; return active->send_rc;
}

static void event(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int bytes) {
    if (type != CHAT_EVT_ATTACHMENT_KEYS_CHANGED) return;
    struct fixture* f = active;
    if (f && !f->allow_result) {
        struct snapshot seen = snapshot(f);
        DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "request test: unexpected event outgoing=%d committed request=%d state=%d command=%d autocommit=%d",
            f->outgoing, seen.exists, seen.state, seen.command, sqlite3_get_autocommit(inst->topo_sqlite_db));
    }
    assert(f && f->inst == inst && f->allow_result && sqlite3_get_autocommit(inst->topo_sqlite_db));
    assert(bytes == (int)strlen(f->channel) + 1 && data[0] == strlen(f->channel) && !memcmp(data + 1, f->channel, strlen(f->channel)));
    equal(f); f->events++;
}

static void receive(struct fixture* f, int success) {
    f->expected = snapshot(f); f->allow_result = success; f->acks = f->events = 0; f->started = database_time(f);
    if (success) {
        if (!f->expected.exists) { f->expected.exists = 1; f->expected.incoming = 1; f->expected.state = 0; f->expected.created = -1; }
        if (f->expected.state) { f->expected.command = f->expected.state == 3 ? 4 : 3; f->expected.retry = 0; }
    }
    struct ll_entry* entry = ll_alloc_lldgram(ROUTER_SVC_PAYLOAD_OFF + 85); assert(entry);
    entry->len = ROUTER_SVC_PAYLOAD_OFF + 85; memset(entry->dgram, 0, entry->len);
    entry->dgram[0] = ETCP_RT_ID_ATTACHMENT_KEYS; entry->dgram[ROUTER_SVC_FLAGS_OFF] = f->flags;
    memcpy(entry->dgram + ROUTER_SVC_SRC_OFF, &f->peer, 8); memcpy(entry->dgram + ROUTER_SVC_DST_OFF, &f->inst->node_id, 8);
    memcpy(entry->dgram + ROUTER_SVC_GROUP_OFF, &f->route.group_id, 8);
    uint8_t* body = entry->dgram + ROUTER_SVC_PAYLOAD_OFF; memcpy(body, "AK01", 4); body[4] = 1;
    memcpy(body + 5, f->message, 32); memcpy(body + 37, f->id, 16); memcpy(body + 53, f->hash, 32);
    if (f->corrupt) body[53] ^= 1;
    struct ETCP_CONN conn = {.instance = f->inst};
    assert(f->inst->router_bindings.callbacks[ETCP_RT_ID_ATTACHMENT_KEYS]);
    f->inst->router_bindings.callbacks[ETCP_RT_ID_ATTACHMENT_KEYS](&conn, entry);
    equal(f); assert(f->events == (unsigned)success && f->acks == (unsigned)(success && f->route_up && f->room));
}

static void reset(struct fixture* f, int state, int queued) {
    char query[768];
    snprintf(query, sizeof(query), "DELETE FROM attachment_key_requests WHERE channel='%s';DELETE FROM attachment_key_outbox WHERE channel='%s'",
        f->channel, f->channel); sql(f, query);
    if (state >= 0) {
        snprintf(query, sizeof(query), "INSERT INTO attachment_key_requests(channel,message_id,peer,incoming,state,created_at)"
            " SELECT channel,message_id,%lld,1,%d,123 FROM attachment_keys WHERE channel='%s'",
            (long long)(sqlite3_int64)f->peer, state, f->channel); sql(f, query);
    }
    if (queued) {
        snprintf(query, sizeof(query), "INSERT INTO attachment_key_outbox(channel,message_id,peer,command,retry_at)"
            " SELECT channel,message_id,peer,%d,17 FROM attachment_key_requests WHERE channel='%s'", state == 3 ? 4 : 3, f->channel);
        sql(f, query);
    }
    f->route_up = f->room = 1; f->send_rc = f->corrupt = 0; f->flags = ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED;
}

static void setup(struct fixture* f) {
    memset(f, 0, sizeof(*f)); strcpy(f->directory, "/tmp/utun-attachment-request-XXXXXX"); assert(mkdtemp(f->directory));
    char config[1024], path[512];
    snprintf(config, sizeof(config), "[global]\ndb_path=%s\n[ntp]\nenabled=no\n"
        "[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\n", f->directory);
    f->ua = uasync_create(); assert(f->ua); f->inst = utun_instance_create_from_str(f->ua, config);
    assert(f->inst && !utun_core_start(f->inst) && !chat_service_start(f->inst));
    chat_core_create_channel_auto(f->inst, "request contract");
    sqlite3_stmt* st = NULL;
    assert(sqlite3_prepare_v2(f->inst->topo_sqlite_db, "SELECT channel_id FROM channels WHERE name='request contract'", -1, &st, NULL) == SQLITE_OK);
    assert(sqlite3_step(st) == SQLITE_ROW); snprintf(f->channel, sizeof(f->channel), "%s", sqlite3_column_text(st, 0));
    assert(sqlite3_finalize(st) == SQLITE_OK);
    uint8_t x[32], ed[32], private_key[32], peer_ed[32], seed[64], join[64], certificate[64], body[112];
    struct SC_MYKEYS peer; assert(sc_generate_keypair(&peer) == SC_OK);
    f->peer = sc_derive_node_id_from_pubkey(peer.public_key); assert(f->peer && f->peer != f->inst->node_id);
    assert(sc_derive_ed25519_pubkey(peer.private_key, peer_ed) == SC_OK); SHA512(peer.private_key, 32, seed);
    assert(topo_node_sqlite_channel_get(f->inst->topo_sqlite_db, f->channel, NULL, 0, NULL, x, ed, NULL) == 0);
    assert(topo_node_sqlite_channel_get_priv(f->inst->topo_sqlite_db, f->channel, private_key) == 0);
    assert(member_sync_build_join_msg(x, ed, f->peer, peer.public_key, 1, body, sizeof(body)) == 112);
    assert(sc_ed25519_sign(seed, body, sizeof(body), join) == SC_OK && member_sync_sign_pubkey(private_key, peer.public_key, certificate) == 0);
    assert(member_sync_put(f->inst, f->channel, f->peer, peer.public_key, peer_ed, join, 1,
        NULL, 0, "{}", NULL, NULL, 0, 0, certificate) >= 0);
    assert(member_sync_verified_ed25519(f->inst, f->channel, f->peer, peer_ed) == 0);
    struct chat_msg_submit request = {.inst = f->inst, .data = (uint8_t*)"request identity", .data_len = 16};
    strcpy(request.channel_id, f->channel); strcpy(request.content_type, "text");
    assert(chat_core_submit_message(f->inst, &request) == 0); memcpy(f->message, request.message_id, 32);
    f->key[0] = 55; f->id[0] = 2; f->hash[0] = 3;
    assert(sqlite3_prepare_v2(f->inst->topo_sqlite_db, "INSERT INTO attachment_keys(channel,crypto_id,message_id,author,key,cipher_hash,cipher_path,plain_path)"
        " VALUES(?,?,?,?,?,?,'','')", -1, &st, NULL) == SQLITE_OK);
    assert(sqlite3_bind_text(st, 1, f->channel, -1, SQLITE_STATIC) == SQLITE_OK);
    assert(sqlite3_bind_blob(st, 2, f->id, 16, SQLITE_STATIC) == SQLITE_OK && sqlite3_bind_blob(st, 3, f->message, 32, SQLITE_STATIC) == SQLITE_OK);
    assert(sqlite3_bind_int64(st, 4, (sqlite3_int64)f->inst->node_id) == SQLITE_OK);
    assert(sqlite3_bind_blob(st, 5, f->key, 32, SQLITE_STATIC) == SQLITE_OK && sqlite3_bind_blob(st, 6, f->hash, 32, SQLITE_STATIC) == SQLITE_OK);
    assert(sqlite3_step(st) == SQLITE_DONE && sqlite3_finalize(st) == SQLITE_OK);
    sql(f, "INSERT INTO attachment_keys(channel,crypto_id,author,key,cipher_path,plain_path,error)"
        " VALUES('900',zeroblob(16),999,x'6300000000000000000000000000000000000000000000000000000000000000','','',7)");
    snprintf(path, sizeof(path), "%s/chats.db", f->directory);
    assert(sqlite3_open_v2(path, &f->reader, SQLITE_OPEN_READONLY, NULL) == SQLITE_OK);
    f->route.inst = f->inst; f->route.group_id = strtoull(f->channel, NULL, 10); f->route.remote_node_id = f->peer;
    f->route.svc_id = ETCP_RT_ID_ATTACHMENT_KEYS; active = f; chat_event_set_handler(f->inst, event); reset(f, -1, 0);
}

static void cleanup(struct fixture* f) {
    chat_event_set_handler(f->inst, NULL); active = NULL; utun_instance_destroy(f->inst);
    assert(sqlite3_close(f->reader) == SQLITE_OK); uasync_destroy(f->ua, 0);
}

static void failed_then_retry(struct fixture* f, int state, int queued, enum failure point, enum operation op, int index) {
    reset(f, state, queued); fault = point; fault_operation = op; fault_index = index; unsigned previous = fault_hits;
    receive(f, 0); assert(!fault && fault_hits == previous + 1 && sqlite3_get_autocommit(f->inst->topo_sqlite_db));
    receive(f, 1); receive(f, 1); cases++;
}

static void test_faults(void) {
    struct fixture f; setup(&f); const int states[] = {-1, 0, 2, 3, 4};
    for (unsigned i = 0; i < sizeof(states) / sizeof(states[0]); i++) for (int queued = 0; queued <= (states[i] >= 2); queued++) {
        int state = states[i]; failed_then_retry(&f, state, queued, BEGIN, NO_OPERATION, 0);
        for (enum operation op = IDENTITY; op <= OUTBOX; op++) {
            if ((op == INSERT && state >= 0) || (op == OUTBOX && state < 2)) continue;
            failed_then_retry(&f, state, queued, PREPARE, op, 0);
            int bindings = op == IDENTITY ? 2 : op == MEMBER ? 1 : op == AUTHOR || op == OUTBOX ? 4 : 3;
            for (int index = 1; index <= bindings; index++) failed_then_retry(&f, state, queued, BIND, op, index);
            failed_then_retry(&f, state, queued, STEP, op, 0); failed_then_retry(&f, state, queued, FINALIZE, op, 0);
            if (op == INSERT || op == OUTBOX) failed_then_retry(&f, state, queued, CHANGES, op, 0);
        }
        failed_then_retry(&f, state, queued, COMMIT, NO_OPERATION, 0);
    }
    cleanup(&f);
    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS attachment requests: %u SQL faults/retries, new and duplicate 0/2/3/4, exact committed snapshots", cases);
}

static int veto(void* arg) { (*(unsigned*)arg)++; return 1; }

static void test_controls(void) {
    struct fixture f; setup(&f); const int states[] = {-1, 0, 2, 3, 4};
    for (unsigned i = 0; i < sizeof(states) / sizeof(states[0]); i++) {
        reset(&f, states[i], 0); sql(&f, "BEGIN;UPDATE attachment_keys SET error=8 WHERE channel='900'");
        receive(&f, 0); assert(!sqlite3_get_autocommit(f.inst->topo_sqlite_db));
        sqlite3_stmt* st = NULL;
        assert(sqlite3_prepare_v2(f.inst->topo_sqlite_db, "SELECT error FROM attachment_keys WHERE channel='900'", -1, &st, NULL) == SQLITE_OK);
        assert(sqlite3_step(st) == SQLITE_ROW && sqlite3_column_int(st, 0) == 8 && sqlite3_finalize(st) == SQLITE_OK);
        sql(&f, "ROLLBACK"); receive(&f, 1);
    }
    const char* triggers[] = {
        "CREATE TEMP TRIGGER reject_request BEFORE INSERT ON attachment_key_requests BEGIN SELECT RAISE(ABORT,'request'); END",
        "CREATE TEMP TRIGGER reject_request BEFORE INSERT ON attachment_key_requests BEGIN SELECT RAISE(IGNORE); END",
        "CREATE TEMP TRIGGER reject_request BEFORE INSERT ON attachment_key_outbox BEGIN SELECT RAISE(ABORT,'queue'); END",
        "CREATE TEMP TRIGGER reject_request BEFORE INSERT ON attachment_key_outbox BEGIN SELECT RAISE(IGNORE); END",
        "CREATE TEMP TRIGGER reject_request BEFORE UPDATE ON attachment_key_outbox BEGIN SELECT RAISE(ABORT,'upsert'); END",
        "CREATE TEMP TRIGGER reject_request BEFORE UPDATE ON attachment_key_outbox BEGIN SELECT RAISE(IGNORE); END"
    };
    for (unsigned i = 0; i < sizeof(triggers) / sizeof(triggers[0]); i++) {
        reset(&f, i < 2 ? -1 : 2, i >= 4); sql(&f, triggers[i]); receive(&f, 0);
        sql(&f, "DROP TRIGGER reject_request"); receive(&f, 1);
    }
    for (int state = -1; state <= 2; state += 3) {
        reset(&f, state, 0); unsigned commits = 0; sqlite3_commit_hook(f.inst->topo_sqlite_db, veto, &commits);
        receive(&f, 0); assert(commits == 1); sqlite3_commit_hook(f.inst->topo_sqlite_db, NULL, NULL); receive(&f, 1);
    }
    for (int state = 0; state <= 5; state++) if (state == 1 || state == 5) { reset(&f, state, 1); receive(&f, 0); }
    reset(&f, 0, 0); char query[256];
    snprintf(query, sizeof(query), "UPDATE attachment_key_requests SET incoming=0 WHERE channel='%s'", f.channel);
    sql(&f, query); receive(&f, 0);
    reset(&f, -1, 0); f.corrupt = 1; receive(&f, 0); f.corrupt = 0;
    f.flags = ROUTER_FLAG_SIGNED; receive(&f, 0); f.flags |= ROUTER_FLAG_ENCRYPTED;
    snprintf(query, sizeof(query), "UPDATE peers_%s SET deleted=1 WHERE node_id=%lld", f.channel, (long long)(sqlite3_int64)f.peer);
    sql(&f, query); receive(&f, 0);
    snprintf(query, sizeof(query), "UPDATE peers_%s SET deleted=0 WHERE node_id=%lld", f.channel, (long long)(sqlite3_int64)f.peer);
    sql(&f, query); receive(&f, 1); cleanup(&f);
    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS attachment requests: foreign transaction preserved, ABORT/IGNORE, commit veto, direction/state/identity/member checks");
}

static void test_ack_retry(void) {
    struct fixture f; setup(&f);
    for (int unavailable = 0; unavailable < 3; unavailable++) {
        reset(&f, -1, 0);
        if (unavailable == 0) f.route_up = 0;
        else if (unavailable == 1) f.room = 0;
        else f.send_rc = -1;
        receive(&f, 1); struct snapshot saved = snapshot(&f);
        f.route_up = f.room = 1; f.send_rc = 0; receive(&f, 1);
        assert(saved.created == f.expected.created && f.expected.command == 0);
    }
    cleanup(&f);
    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS attachment requests: missing route, backpressure and send failure retain request; duplicate retries ACK without new consent");
}

static void test_rollback_failure(void) {
    struct fixture f; setup(&f); fault = COMMIT; fault_operation = NO_OPERATION; fail_rollback = 1; receive(&f, 0);
    assert(!fault && !fail_rollback && !f.inst->running && !sqlite3_get_autocommit(f.inst->topo_sqlite_db));
    assert(!f.inst->router_bindings.callbacks[ETCP_RT_ID_ATTACHMENT_KEYS]);
    chat_event_set_handler(f.inst, NULL); active = NULL; utun_instance_destroy(f.inst);
    equal(&f); assert(sqlite3_close(f.reader) == SQLITE_OK); uasync_destroy(f.ua, 0);
    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS attachment requests: failed rollback sends no ACK/event, quiesces/stops core, owner close rolls back request");
}

/* Получатель уже имеет проверенный ciphertext; именно эти локальные строки потребляет таймер. */
static void outgoing_reset(struct fixture* f, int queued) {
    reset(f, -1, 0); f->outgoing = 1;
    char query[768];
    snprintf(query, sizeof(query), "UPDATE attachment_keys SET author=%lld,key=NULL,downloaded=1,wanted=1,ready=0,decrypting=0,error=0"
        " WHERE channel='%s'", (long long)(sqlite3_int64)f->peer, f->channel); sql(f, query);
    if (queued) {
        snprintf(query, sizeof(query), "INSERT INTO attachment_key_outbox(channel,message_id,peer,command,retry_at)"
            " SELECT channel,message_id,author,4,9223372036854775807 FROM attachment_keys WHERE channel='%s'", f->channel); sql(f, query);
    }
}

static void outgoing_tick(struct fixture* f, int create, int sends) {
    f->expected = snapshot(f); f->allow_result = create || sends; f->acks = f->events = 0; f->started = database_time(f);
    if (create) {
        f->expected.exists = 1; f->expected.incoming = 0; f->expected.state = 0;
        f->expected.created = -1; f->expected.command = 1; f->expected.retry = 0;
    }
    assert(attachment_tick && CC(f->inst)->attachment_timer);
    uasync_cancel_timeout(f->ua, CC(f->inst)->attachment_timer); attachment_tick(f->inst);
    if (sends || (create && (!f->route_up || !f->room))) {
        f->expected.retry = snapshot(f).retry;
        assert(f->expected.retry >= f->started + 5 && f->expected.retry <= database_time(f) + 5);
    }
    equal(f); assert(f->events == (unsigned)create && f->acks == (unsigned)sends);
}

static void test_outgoing_controls(void) {
    struct fixture f; setup(&f); outgoing_reset(&f, 0);
    const char* triggers[] = {
        "CREATE TEMP TRIGGER reject_request BEFORE INSERT ON attachment_key_requests BEGIN SELECT RAISE(IGNORE); END",
        "CREATE TEMP TRIGGER reject_request BEFORE INSERT ON attachment_key_requests BEGIN SELECT RAISE(ABORT,'request'); END",
        "CREATE TEMP TRIGGER reject_request BEFORE INSERT ON attachment_key_outbox BEGIN SELECT RAISE(IGNORE); END",
        "CREATE TEMP TRIGGER reject_request BEFORE INSERT ON attachment_key_outbox BEGIN SELECT RAISE(ABORT,'outbox'); END",
        "CREATE TEMP TRIGGER reject_request BEFORE UPDATE ON attachment_key_outbox BEGIN SELECT RAISE(IGNORE); END",
        "CREATE TEMP TRIGGER reject_request BEFORE UPDATE ON attachment_key_outbox BEGIN SELECT RAISE(ABORT,'upsert'); END"
    };
    for (unsigned i = 0; i < sizeof(triggers) / sizeof(triggers[0]); i++) {
        outgoing_reset(&f, i >= 4); sql(&f, triggers[i]); outgoing_tick(&f, 0, 0);
        sql(&f, "DROP TRIGGER reject_request"); outgoing_tick(&f, 1, 1); outgoing_tick(&f, 0, 0);
    }
    outgoing_reset(&f, 0); sql(&f, "BEGIN;UPDATE attachment_keys SET error=8 WHERE channel='900'");
    outgoing_tick(&f, 0, 0); assert(!sqlite3_get_autocommit(f.inst->topo_sqlite_db));
    sql(&f, "ROLLBACK"); outgoing_tick(&f, 1, 1);
    /* Уже сохранённая очередь во внешней транзакции также не уходит в сеть. */
    sql(&f, "UPDATE attachment_key_outbox SET retry_at=0;BEGIN;UPDATE attachment_keys SET error=8 WHERE channel='900'");
    outgoing_tick(&f, 0, 0); assert(!sqlite3_get_autocommit(f.inst->topo_sqlite_db));
    sql(&f, "ROLLBACK"); outgoing_tick(&f, 0, 1);
    outgoing_reset(&f, 0); unsigned commits = 0; sqlite3_commit_hook(f.inst->topo_sqlite_db, veto, &commits);
    outgoing_tick(&f, 0, 0); assert(commits == 1); sqlite3_commit_hook(f.inst->topo_sqlite_db, NULL, NULL);
    outgoing_tick(&f, 1, 1); cleanup(&f);
    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS outgoing requests: ABORT/IGNORE, commit veto, foreign transaction postpones creation and send");
}

static void test_outgoing_faults(void) {
    struct fixture f; setup(&f); unsigned count = 0;
    for (int queued = 0; queued < 2; queued++) {
        const enum operation operations[] = {NO_OPERATION, PENDING, ELIGIBLE, INSERT, OUTBOX, NO_OPERATION};
        for (unsigned i = 0; i < sizeof(operations) / sizeof(operations[0]); i++) {
            enum operation op = operations[i];
            int binds = op == ELIGIBLE || op == INSERT ? 3 : op == OUTBOX ? 4 : 0;
            int writes = op == INSERT || op == OUTBOX;
            int points = op ? 3 + binds + writes : 1;
            for (int point = 0; point < points; point++) {
                outgoing_reset(&f, queued);
                fault = !op ? (i ? COMMIT : BEGIN) : point == 0 ? PREPARE : point <= binds ? BIND :
                    point == binds + 1 ? STEP : point == binds + 2 ? FINALIZE : CHANGES;
                fault_operation = op; fault_index = point; unsigned previous = fault_hits;
                outgoing_tick(&f, 0, 0); assert(!fault && fault_hits == previous + 1 && sqlite3_get_autocommit(f.inst->topo_sqlite_db));
                outgoing_tick(&f, 1, 1); outgoing_tick(&f, 0, 0); count++;
            }
        }
    }
    cleanup(&f);
    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS outgoing requests: %u SQL faults/retries, both outbox paths, exact committed snapshots", count);
}

static void test_outgoing_rollback_failure(void) {
    struct fixture f; setup(&f); outgoing_reset(&f, 0);
    fault = COMMIT; fault_operation = NO_OPERATION; fail_rollback = 1; outgoing_tick(&f, 0, 0);
    assert(!fault && !fail_rollback && !f.inst->running && !sqlite3_get_autocommit(f.inst->topo_sqlite_db));
    assert(!CC(f.inst)->attachment_timer && !f.inst->router_bindings.callbacks[ETCP_RT_ID_ATTACHMENT_KEYS]);
    chat_event_set_handler(f.inst, NULL); active = NULL; utun_instance_destroy(f.inst);
    equal(&f); assert(sqlite3_close(f.reader) == SQLITE_OK); uasync_destroy(f.ua, 0);
    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS outgoing requests: rollback failure stops core without event/send or rearming timer");
}

static void test_outgoing_restart(void) {
    struct fixture f; setup(&f);
    for (int unavailable = 0; unavailable < 3; unavailable++) {
        outgoing_reset(&f, 0);
        if (!unavailable) f.route_up = 0;
        else if (unavailable == 1) f.room = 0;
        else f.send_rc = -1;
        outgoing_tick(&f, 1, unavailable == 2);
        struct snapshot saved = snapshot(&f);
        chat_event_set_handler(f.inst, NULL); active = NULL; utun_instance_destroy(f.inst);
        char config[1024];
        snprintf(config, sizeof(config), "[global]\ndb_path=%s\n[ntp]\nenabled=no\n"
            "[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\n", f.directory);
        f.inst = utun_instance_create_from_str(f.ua, config);
        assert(f.inst && !utun_core_start(f.inst) && !chat_service_start(f.inst));
        f.route.inst = f.inst; f.route_up = f.room = 1; f.send_rc = 0; active = &f;
        chat_event_set_handler(f.inst, event);
        struct snapshot restored = snapshot(&f);
        assert(restored.exists && restored.incoming == 0 && restored.state == 0 && restored.command == 1 && restored.retry == 0);
        assert(restored.created == saved.created); outgoing_tick(&f, 0, 1);
    }
    cleanup(&f);
    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS outgoing requests: no route/backpressure/send error preserve queue and date across real instance restart");
}

static void pump_reset(struct fixture* f) {
    outgoing_reset(f, 0); f->pump_only = 1;
    char query[768];
    snprintf(query, sizeof(query), "INSERT INTO attachment_key_requests(channel,message_id,peer,incoming,state,created_at)"
        " SELECT channel,message_id,author,0,0,123 FROM attachment_keys WHERE channel='%s';"
        " INSERT INTO attachment_key_outbox(channel,message_id,peer,command,retry_at)"
        " SELECT channel,message_id,author,1,0 FROM attachment_keys WHERE channel='%s'", f->channel, f->channel); sql(f, query);
}

static void test_pump(void) {
    struct fixture f; setup(&f); unsigned count = 0;
    const enum operation operations[] = {NO_OPERATION, PUMP_READ, PUMP_RETRY, NO_OPERATION};
    for (unsigned op = 0; op < sizeof(operations) / sizeof(operations[0]); op++) {
        int bindings = operations[op] == PUMP_RETRY ? 3 : 0, points = operations[op] ? 3 + bindings + (bindings != 0) : 1;
        for (int point = 0; point < points; point++) {
            pump_reset(&f); fault_operation = operations[op]; fault_index = point;
            fault = !operations[op] ? (op ? COMMIT : BEGIN) : point == 0 ? PREPARE : point <= bindings ? BIND :
                point == bindings + 1 ? STEP : point == bindings + 2 ? FINALIZE : CHANGES;
            unsigned previous = fault_hits; outgoing_tick(&f, 0, 0);
            assert(!fault && fault_hits == previous + 1 && sqlite3_get_autocommit(f.inst->topo_sqlite_db));
            outgoing_tick(&f, 0, 1); outgoing_tick(&f, 0, 0); count++;
        }
    }
    const char* triggers[] = {
        "CREATE TEMP TRIGGER reject_retry BEFORE UPDATE ON attachment_key_outbox BEGIN SELECT RAISE(IGNORE); END",
        "CREATE TEMP TRIGGER reject_retry BEFORE UPDATE ON attachment_key_outbox BEGIN SELECT RAISE(ABORT,'retry'); END"
    };
    for (unsigned i = 0; i < 2; i++) {
        pump_reset(&f); sql(&f, triggers[i]); outgoing_tick(&f, 0, 0);
        sql(&f, "DROP TRIGGER reject_retry"); outgoing_tick(&f, 0, 1);
    }
    pump_reset(&f); unsigned commits = 0; sqlite3_commit_hook(f.inst->topo_sqlite_db, veto, &commits);
    outgoing_tick(&f, 0, 0); assert(commits == 1); sqlite3_commit_hook(f.inst->topo_sqlite_db, NULL, NULL); outgoing_tick(&f, 0, 1);
    pump_reset(&f); fault = COMMIT; fault_operation = NO_OPERATION; fail_rollback = 1; outgoing_tick(&f, 0, 0);
    assert(!fault && !fail_rollback && !f.inst->running && !CC(f.inst)->attachment_timer && !sqlite3_get_autocommit(f.inst->topo_sqlite_db));
    chat_event_set_handler(f.inst, NULL); active = NULL; utun_instance_destroy(f.inst); equal(&f);
    assert(sqlite3_close(f.reader) == SQLITE_OK); uasync_destroy(f.ua, 0);
    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS outbox:%u SQL faults, ABORT/IGNORE/veto/rollback failure; no send before retry commit, exact queue/date preserved", count);
}

int main(int argc, char** argv) {
    debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_MEDIA, DEBUG_LEVEL_INFO);
    utun_instance_set_tun_init_enabled(0); size_t baseline = u_get_allocated_count();
    if (argc != 2 || strcmp(argv[1], "--outgoing")) { test_faults(); test_controls(); test_ack_retry(); test_rollback_failure(); }
    test_outgoing_controls(); test_outgoing_faults(); test_outgoing_restart(); test_outgoing_rollback_failure();
    test_pump();
    assert(!fault && !fail_rollback && u_get_allocated_count() == baseline);
    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS attachment requests: reader before ACK/events, stable duplicate date, neighboring key and allocations preserved");
    return 0;
}