/* Собственная SQLite и настоящие подписи: сохранность истории, очистка порчи,
 * транзакционные отказы, восстановление цепочки и точная идентичность ACK. */
#include "../src/chat/db_sync.c"
#include <assert.h>

static int fail_verify, fail_commit, fail_timer, fail_blob, blob_nomem;
static unsigned fault_hits;
static sqlite3* verification_db;
static int assert_unlocked;
static int concentrate_bucket;
extern int __real_chat_message_id(uint64_t, uint64_t, const uint8_t[64], uint8_t[32]);
int __wrap_chat_message_id(uint64_t group, uint64_t ts, const uint8_t sig[64], uint8_t key[32]) {
    int rc = __real_chat_message_id(group, ts, sig, key);
    /* Только для плотного тестового листа: подписи настоящие, оставшиеся223 бита ключа сохраняются. */
    if (rc == 0 && concentrate_bucket) { key[0] = key[1] = key[2] = 0; key[3] &= 0x7f; }
    return rc;
}
extern int __real_EVP_DigestVerifyInit(EVP_MD_CTX*, EVP_PKEY_CTX**, const EVP_MD*, ENGINE*, EVP_PKEY*);
int __wrap_EVP_DigestVerifyInit(EVP_MD_CTX* ctx, EVP_PKEY_CTX** pctx, const EVP_MD* md, ENGINE* engine, EVP_PKEY* key) {
    if (assert_unlocked) assert(sqlite3_get_autocommit(verification_db));
    if (fail_verify && !--fail_verify) { fault_hits++; return 0; }
    return __real_EVP_DigestVerifyInit(ctx, pctx, md, engine, key);
}

extern int __real_sqlite3_exec(sqlite3*, const char*, int (*)(void*, int, char**, char**), void*, char**);
int __wrap_sqlite3_exec(sqlite3* db, const char* sql, int (*cb)(void*, int, char**, char**), void* arg, char** err) {
    if (fail_commit && !strcmp(sql, "COMMIT")) { fail_commit = 0; fault_hits++; return SQLITE_BUSY; }
    return __real_sqlite3_exec(db, sql, cb, arg, err);
}

extern const void* __real_sqlite3_column_blob(sqlite3_stmt*, int);
const void* __wrap_sqlite3_column_blob(sqlite3_stmt* stmt, int col) {
    if (fail_blob) { fail_blob = 0; blob_nomem = 1; fault_hits++; return NULL; }
    return __real_sqlite3_column_blob(stmt, col);
}
extern int __real_sqlite3_errcode(sqlite3*);
int __wrap_sqlite3_errcode(sqlite3* db) {
    if (blob_nomem) { blob_nomem = 0; return SQLITE_NOMEM; }
    return __real_sqlite3_errcode(db);
}

extern void* __real_uasync_set_timeout(struct UASYNC*, uint64_t, void*, void (*)(void*), const char*);
void* __wrap_uasync_set_timeout(struct UASYNC* ua, uint64_t delay, void* arg, void (*cb)(void*), const char* name) {
    if (fail_timer && !strcmp(name, "history_preflight")) { fail_timer = 0; fault_hits++; return NULL; }
    return __real_uasync_set_timeout(ua, delay, arg, cb, name);
}


struct fixture {
    struct UTUN_INSTANCE inst;
    struct DB_SYNC db;
    struct DB_SYNC_INSTANCE* si;
    struct DB_SYNC_INSTANCE* other;
    uint8_t seed[32];
};
static int empty_hash(void* ctx, const char* ns, uint8_t level, uint64_t prefix, EVP_MD_CTX* hash) {
    (void)ctx; (void)ns; (void)level; (void)prefix; (void)hash; return 0;
}
static int empty_page(void* ctx, const char* ns, uint64_t prefix, int av, const merkle_key_t* after,
        uint8_t* buf, size_t* len, merkle_key_t* next, int* more) {
    (void)ctx; (void)ns; (void)prefix; (void)av; (void)after; (void)buf;
    *len = 0; *more = 0; memset(next, 0, sizeof(*next)); return 0;
}
static int empty_apply(void* ctx, const char* ns, uint64_t peer, const uint8_t* p, size_t len) {
    (void)ctx; (void)ns; (void)peer; (void)p; return len ? -1 : 0;
}
static const struct merkle_sync_data_ops empty_ops = {
    .update_bucket_hash = empty_hash, .get_page = empty_page, .apply_items = empty_apply
};
static void sql(struct fixture* f, const char* query) {
    int rc = sqlite3_exec(f->db.db, query, NULL, NULL, NULL);
    if (rc != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "integrity SQL %s: %s", query, sqlite3_errmsg(f->db.db));
    assert(rc == SQLITE_OK);
}
static struct DB_SYNC_INSTANCE* new_table(struct fixture* f, const char* name, uint64_t group) {
    struct DB_SYNC_INSTANCE* si = u_calloc(1, sizeof(*si)); assert(si);
    si->db_sync = &f->db; si->enabled = 1; si->group_id = group; si->next_id = 1; si->sync_gated = 1;
    snprintf(si->table_name, sizeof(si->table_name), "%s", name);
    snprintf(si->ns, sizeof(si->ns), "%llu", (unsigned long long)group);
    assert(si_table_create(si, name, 0) == 0); return si;
}
static void init(struct fixture* f) {
    memset(f, 0, sizeof(*f)); f->inst.node_id = 1; f->inst.ua = uasync_create(); assert(f->inst.ua);
    f->inst.connections = queue_new(f->inst.ua, 8, 0, 8, "history test"); assert(f->inst.connections);
    memset(f->seed, 9, sizeof(f->seed));
    EVP_PKEY* key = EVP_PKEY_new_raw_private_key(EVP_PKEY_ED25519, NULL, f->seed, 32); assert(key);
    size_t len = 32;
    assert(EVP_PKEY_get_raw_public_key(key, f->inst.my_ed25519_pubkey, &len) == 1); EVP_PKEY_free(key);
    assert(sqlite3_open(":memory:", &f->db.db) == SQLITE_OK);
    f->db.inst = &f->inst; f->db.enabled = 1; f->inst.db_sync = &f->db; f->inst.topo_sqlite_db = f->db.db;
    verification_db = f->db.db;
    sql(f, "CREATE TABLE nodes(node_id INTEGER PRIMARY KEY,ed25519_pubkey BLOB,deleted INTEGER DEFAULT 0)");
    f->si = new_table(f, "history", 10); f->other = new_table(f, "other_history", 11);
    f->db.instances = u_calloc(2, sizeof(*f->db.instances)); assert(f->db.instances);
    f->db.instances[0] = f->si; f->db.instances[1] = f->other; f->db.instance_count = f->db.instance_capacity = 2;
    assert(merkle_sync_init(&f->inst, ETCP_ID_MEMBER_SYNC, &empty_ops, f) == 0 && db_sync_register_model(&f->inst) == 0);
}
static void destroy(struct fixture* f) {
    f->db.enabled = 0;
    if (f->db.peer_check_timer) uasync_cancel_timeout(f->inst.ua, f->db.peer_check_timer);
    merkle_sync_destroy(&f->inst);
    while (f->db.instance_count) db_sync_instance_remove(f->db.instances[f->db.instance_count - 1]);
    while (f->db.done_cbks) { struct db_sync_done_cbk_entry* cb = f->db.done_cbks; f->db.done_cbks = cb->next; u_free(cb); }
    u_free(f->db.instances); queue_free(f->inst.connections);
    assert(sqlite3_get_autocommit(f->db.db) && sqlite3_close(f->db.db) == SQLITE_OK);
    uasync_destroy(f->inst.ua, 0); verification_db = NULL;
}
static void sign(struct fixture* f, uint64_t ts, const char* text, uint8_t sig[64]) {
    uint8_t body[DB_SYNC_MESSAGE_MAX + 8]; size_t len = strlen(text); assert(len <= DB_SYNC_MESSAGE_MAX);
    memcpy(body, &ts, 8); memcpy(body + 8, text, len); assert(sc_ed25519_sign(f->seed, body, len + 8, sig) == SC_OK);
}
static int insert(struct fixture* f, uint64_t ts, const char* text, const uint8_t sig[64]) {
    return db_sync_insert_signed(f->si, text, strlen(text), sig, 64, ts, NULL);
}
static void intact(struct fixture* f, const char* text) {
    sqlite3_stmt* st = NULL;
    assert(sqlite3_prepare_v2(f->db.db, "SELECT data FROM history", -1, &st, NULL) == SQLITE_OK && sqlite3_step(st) == SQLITE_ROW);
    assert(sqlite3_column_bytes(st, 0) == (int)strlen(text) && !memcmp(sqlite3_column_blob(st, 0), text, strlen(text)));
    assert(sqlite3_step(st) == SQLITE_DONE && sqlite3_finalize(st) == SQLITE_OK && sqlite3_get_autocommit(f->db.db));
}
static int page(struct fixture* f, const uint8_t sig[64]) {
    uint8_t id[32], data[16384]; size_t len = sizeof(data); merkle_key_t after = {{0}}, next; int more;
    assert(chat_message_id(10, 100, sig, id) == 0);
    return db_message_page(&f->inst, "10", merkle_sync_level_prefix(db_read64(id), MT_MAX_LEVEL), 0, &after, data, &len, &next, &more);
}
static void test_preservation(void) {
    struct fixture f; init(&f); uint8_t sig[64]; sign(&f, 100, "original", sig);
    assert(insert(&f, 100, "original", sig) == 0 && db_sync_verify(f.si) == 0);
    assert(insert(&f, 100, "original", sig) == 1);
    fail_verify = 1; assert(insert(&f, 100, "original", sig) == -1); intact(&f, "original");
    fail_verify = 2; assert(insert(&f, 100, "modified", sig) == -1); intact(&f, "original");
    fail_blob = 1; assert(insert(&f, 100, "original", sig) == -1); intact(&f, "original");
    fail_blob = 1; assert(page(&f, sig) == -1); intact(&f, "original");
    fail_verify = 1; assert(page(&f, sig) == -1); intact(&f, "original");
    assert(insert(&f, 100, "modified", sig) == -2); intact(&f, "original");
    sql(&f, "UPDATE history SET node_id=99");
    assert(insert(&f, 100, "original", sig) == -1 && page(&f, sig) == -1); intact(&f, "original");
    assert(db_sync_verify(f.si) == 1 && f.si->index.pending);
    sql(&f, "DROP TABLE nodes");
    assert(db_sync_verify(f.si) == -1); intact(&f, "original");
    sql(&f, "CREATE TABLE nodes(node_id INTEGER PRIMARY KEY,ed25519_pubkey BLOB,deleted INTEGER);UPDATE history SET node_id=1");
    assert(db_sync_verify(f.si) == 0);
    destroy(&f);
    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS history valid copy survives invalid input, crypto/SQL/BLOB failures and missing key");
}
static void test_deletion(void) {
    struct fixture f; init(&f); uint8_t sig[64]; sign(&f, 100, "original", sig);
    assert(insert(&f, 100, "original", sig) == 0);
    assert(db_sync_insert_signed(f.other, "original", 8, sig, 64, 100, NULL) == 0 && db_sync_verify(f.si) == 0);
    sql(&f, "UPDATE history SET data='modified';CREATE TRIGGER deny_delete BEFORE DELETE ON history BEGIN SELECT RAISE(FAIL,'delete denied');END");
    assert(insert(&f, 100, "modified", sig) == -1); intact(&f, "modified");
    sql(&f, "DROP TRIGGER deny_delete");
    fail_commit = 1; assert(insert(&f, 100, "modified", sig) == -1); intact(&f, "modified");
    assert(insert(&f, 100, "modified", sig) == -2 && db_sync_count(f.si) == 0 && f.si->index.pending);
    assert(db_sync_verify(f.si) == 0 && db_sync_count(f.other) == 1);
    assert(insert(&f, 100, "original", sig) == 0); sql(&f, "UPDATE history SET data='modified'");
    assert(insert(&f, 100, "original", sig) == 0); intact(&f, "original");
    sql(&f, "UPDATE history SET data=zeroblob(4001)");
    assert(db_sync_verify(f.si) == 0 && db_sync_count(f.si) == 0);
    const unsigned sizes[] = {0,1,63,65,255};
    for (unsigned i = 0; i < sizeof(sizes) / sizeof(*sizes); i++) {
        assert(insert(&f, 100, "original", sig) == 0);
        char query[96]; snprintf(query, sizeof(query), "UPDATE history SET author_signature=zeroblob(%u)", sizes[i]); sql(&f, query);
        assert(db_sync_verify(f.si) == 0 && db_sync_count(f.si) == 0);
    }
    destroy(&f);
    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS proven malformed/signature-invalid stored records removed; rollback preserves data and other channel");
}
static void test_index(void) {
    struct fixture f; init(&f); uint8_t sig[64], root[32], prior[32];
    for (uint64_t ts = 1; ts <= 55; ts++) { sign(&f, ts, "record", sig); assert(insert(&f, ts, "record", sig) == 0); }
    assert_unlocked = 1; fail_blob = 1;
    assert(db_sync_verify(f.si) == -1 && db_sync_count(f.si) == 55 && f.si->index.pending);
    fail_commit = 1; assert(db_sync_verify(f.si) == -1 && f.si->index.pending && sqlite3_get_autocommit(f.db.db));
    sql(&f, "CREATE TRIGGER deny_hash BEFORE INSERT ON merkle_tree_hash WHEN NEW.namespace='2:10' BEGIN SELECT RAISE(FAIL,'hash denied');END");
    assert(db_sync_verify(f.si) == -1 && f.si->index.pending && db_sync_read_root(f.si, root) == 1);
    sql(&f, "DROP TRIGGER deny_hash");
    assert(db_sync_verify(f.si) == 0 && db_sync_read_root(f.si, prior) == 0); assert_unlocked = 0;
    uint64_t next = f.si->next_id;
    sign(&f, 56, "record", sig); fail_commit = 1;
    assert(insert(&f, 56, "record", sig) == -1 && db_sync_count(f.si) == 55 && f.si->next_id == next);
    assert(db_sync_read_root(f.si, root) == 0 && !memcmp(root, prior, 32));
    fail_timer = 1; db_index_request(f.si);
    assert(!fail_timer && f.si->index.pending && !f.si->index.timer && db_sync_verify(f.si) == 0);
    assert(db_sync_read_root(f.si, prior) == 0);
    sql(&f, "UPDATE history SET id=id+900,flags=1,local_attrs='local',delivered_peers=17,delivery_chain='different'");
    assert(db_sync_verify(f.si) == 0 && db_sync_read_root(f.si, root) == 0 && !memcmp(root, prior, 32));
    destroy(&f);
    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS bounded preflight verifies outside write-lock; index failures recover; root ignores local ids and delivery");
}
static void test_delivery(void) {
    struct fixture f; init(&f); uint8_t a[64], b[64], id_a[32], id_b[32];
    sign(&f, 100, "first", a); sign(&f, 100, "second", b);
    assert(insert(&f, 100, "first", a) == 0 && insert(&f, 100, "second", b) == 0);
    assert(chat_message_id(10, 100, a, id_a) == 0 && chat_message_id(10, 100, b, id_b) == 0);
    assert(si_delivery_update(f.si, id_a, 2) == 0 && si_delivery_update(f.si, id_a, 2) == 0);
    sqlite3_stmt* st = NULL;
    assert(sqlite3_prepare_v2(f.db.db, "SELECT data,delivered_peers,flags FROM history ORDER BY data", -1, &st, NULL) == SQLITE_OK);
    assert(sqlite3_step(st) == SQLITE_ROW && sqlite3_column_int(st, 1) == 1 && sqlite3_column_int(st, 2) == 1);
    assert(sqlite3_step(st) == SQLITE_ROW && !sqlite3_column_int(st, 1) && !sqlite3_column_int(st, 2));
    assert(sqlite3_step(st) == SQLITE_DONE && sqlite3_finalize(st) == SQLITE_OK);
    sql(&f, "CREATE TRIGGER deny_ack BEFORE UPDATE ON history BEGIN SELECT RAISE(FAIL,'ACK denied');END");
    assert(si_delivery_update(f.si, id_b, 3) == -1); sql(&f, "DROP TRIGGER deny_ack");
    assert(si_delivery_update(f.si, id_b, 3) == 0);
    destroy(&f);
    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS exact same-timestamp ACK identity and idempotence");
}
static int first_calls, second_calls;
static void second_done(struct DB_SYNC_INSTANCE* si, uint64_t peer, void* arg) {
    (void)si; (void)peer; (void)arg; second_calls++;
}
static void first_done(struct DB_SYNC_INSTANCE* si, uint64_t peer, void* arg) {
    (void)peer; struct fixture* f = arg; first_calls++;
    db_sync_remove_done_cbk(&f->inst, first_done, f); db_sync_remove_done_cbk(&f->inst, second_done, f);
    db_sync_instance_remove(si);
}
static void test_callbacks(void) {
    struct fixture f; init(&f);
    db_sync_add_done_cbk(&f.inst, second_done, &f); db_sync_add_done_cbk(&f.inst, first_done, &f);
    db_message_result(&f.inst, "10", 2, MT_OK);
    assert(first_calls == 1 && !second_calls && !db_instance_find(&f.db, 10) && !f.db.done_cbks);
    destroy(&f);
    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS completion can remove itself, next callback and channel safely");
}

static void copy_pages(struct fixture* source, struct fixture* target, int insert_before_cursor) {
    merkle_key_t after = {{0}}, next; int av = 0, more, pages = 0;
    do {
        uint8_t data[16384]; size_t len = sizeof(data);
        assert(db_message_page(&source->inst, "10", 0, av, &after, data, &len, &next, &more) == 0);
        assert(db_message_apply(&target->inst, "10", 2, 0, av, &after, &next, more, data, len) == 0);
        after = next; av = 1; pages++;
        if (pages == 1 && insert_before_cursor) {
            uint8_t sig[64], key[32]; int found = 0;
            for (uint64_t ts = 1000; ts < 2000; ts++) {
                sign(source, ts, "inserted before cursor", sig);
                assert(chat_message_id(10, ts, sig, key) == 0);
                if (memcmp(key, after.bytes, 32) < 0) {
                    assert(insert(source, ts, "inserted before cursor", sig) == 0); found = 1; break;
                }
            }
            assert(found);
        }
    } while (more);
    assert(pages >= 3);
}
static void test_pages(void) {
    concentrate_bucket = 1;
    struct fixture a, b; init(&a); init(&b); uint8_t sig[64], ra[32], rb[32];
    b.si->next_id = 900;
    for (uint64_t ts = 1; ts <= 40; ts++) { sign(&a, ts, "page record", sig); assert(insert(&a, ts, "page record", sig) == 0); }
    assert(db_sync_verify(a.si) == 0 && db_sync_verify(b.si) == 0);
    copy_pages(&a, &b, 1);
    assert(db_sync_count(a.si) == 41 && db_sync_count(b.si) == 40);
    assert(db_sync_read_root(a.si, ra) == 0 && db_sync_read_root(b.si, rb) == 0 && memcmp(ra, rb, 32));
    copy_pages(&a, &b, 0);
    assert(db_sync_count(b.si) == 41 && db_sync_read_root(b.si, rb) == 0 && !memcmp(ra, rb, 32));
    copy_pages(&a, &b, 0); assert(db_sync_count(b.si) == 41);
    destroy(&a); destroy(&b); concentrate_bucket = 0;
    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS multi-page leaf, different local ids, duplicate replay and insertion before cursor need a new root check");
}
static void test_deleted_author(void) {
    struct fixture f; init(&f); uint8_t sig[64]; sign(&f, 100, "historic", sig);
    struct db_record r = {.timestamp = 100, .author = 99, .data = "historic", .data_len = 8, .sig_len = 64};
    memcpy(r.signature, sig, 64);
    sqlite3_stmt* st = NULL;
    assert(sqlite3_prepare_v2(f.db.db, "INSERT INTO nodes VALUES(99,?,1)", -1, &st, NULL) == SQLITE_OK);
    assert(sqlite3_bind_blob(st, 1, f.inst.my_ed25519_pubkey, 32, SQLITE_STATIC) == SQLITE_OK && sqlite3_step(st) == SQLITE_DONE);
    assert(sqlite3_finalize(st) == SQLITE_OK);
    sql(&f, "CREATE TABLE peers_10(node_id INTEGER PRIMARY KEY,deleted INTEGER);INSERT INTO peers_10 VALUES(99,1)");
    assert(db_insert_one(f.si, &r, NULL) == 0 && db_sync_verify(f.si) == 0 && page(&f, sig) == 0 && db_sync_count(f.si) == 1);
    destroy(&f);
    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS valid history of a deleted author remains in the root and outgoing page");
}
static void test_ttl(void) {
    struct fixture f; init(&f); uint8_t sig[64];
    struct utun_config* config = u_calloc(1, sizeof(*config)); assert(config); f.inst.config = config;
    config->global.db_sync_ttl = 3600;
    uint64_t now = db_now_ms(f.si), old = now - 3610000;
    sign(&f, now, "fresh", sig); assert(insert(&f, now, "fresh", sig) == 0);
    for (unsigned i = 0; i < 20; i++) { sign(&f, old + i, "expired", sig); assert(insert(&f, old + i, "expired", sig) == 0); }
    sql(&f, "UPDATE history SET flags=1 WHERE rowid=2");
    assert(db_sync_verify(f.si) == 0);
    db_ttl_tick(f.si);
    assert(db_sync_count(f.si) == 5 && f.si->index.pending && sqlite3_get_autocommit(f.db.db));
    uint64_t deadline = get_time_tb() + 10000;
    while (db_sync_count(f.si) != 2 && get_time_tb() < deadline) uasync_poll(f.inst.ua, 1);
    assert(db_sync_count(f.si) == 2 && db_sync_verify(f.si) == 0);
    destroy(&f); u_free(config);
    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS TTL uses milliseconds and short batches, preserving fresh and individually acknowledged records");
}
int main(void) {
    debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_CHAT_SYNC, DEBUG_LEVEL_DEBUG);
    size_t baseline = u_get_allocated_count();
    test_preservation(); test_deletion(); test_index(); test_delivery(); test_callbacks();
    test_pages(); test_deleted_author(); test_ttl();
    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history injected failures=%u allocations=%zu baseline=%zu", fault_hits, u_get_allocated_count(), baseline);
    assert(fault_hits == 10 && u_get_allocated_count() == baseline);
    return 0;
}