// SPDX-License-Identifier: GPL-3.0-or-later
//
// This file is part of CProof.
// See LICENSE for the full GPLv3 text and the special OpenSSL linking exception.

/*
 * bench_export_import.c
 * vim: expandtab:ts=4:sts=4:sw=4
 *
 * S7/S8 export + import bench harness.
 *
 * Subcommands:
 *   seed    --db=PATH --rows=N [--profile=mixed] [--lmc-rate=PCT]
 *   export  --account=JID --csv=PATH [--label=TAG]
 *   import  --account=JID --csv=PATH [--label=TAG]
 *   roundtrip --rows=N --csv=PATH                 # seed → export → fresh DB → import → full diff
 *   verify  --db-a=PATH --db-b=PATH               # full byte-by-byte content diff of two SQLite DBs
 *
 * Layout (under $BENCH_DATA_DIR, default /tmp/cproof-bench-export):
 *   database/<account_jid_dir>/chatlog.db
 *   flatlog/<account_jid_dir>/<contact_jid_dir>/history.log
 *
 * Seeding writes directly to SQLite via a separate sqlite3 handle (faster than
 * dispatching through the backend's add_incoming for each row). Export/import
 * uses the *real* log_database_export_to_flatfile / _import_from_flatfile.
 */

#include "config.h"

#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <time.h>
#include <unistd.h>

#include <glib.h>
#include <glib/gstdio.h>
#include <sqlite3.h>

#include "bench_common.h"
#include "bench_csv.h"
#include "config/account.h"
#include "database.h"
#include "database_flatfile.h"

// ---------------------------------------------------------------------------
// Tiny RNG / body helpers — duplicated from gen_history (kept local so this
// binary is independent of gen_history layout).

static uint64_t
xrng(uint64_t* s)
{
    uint64_t x = *s;
    x ^= x << 13;
    x ^= x >> 7;
    x ^= x << 17;
    *s = x;
    return x;
}

static char*
make_uuid(uint64_t* rng)
{
    uint64_t a = xrng(rng), b = xrng(rng);
    return g_strdup_printf("%08x-%04x-4%03x-%04x-%012lx",
                           (unsigned)(a >> 32),
                           (unsigned)((a >> 16) & 0xffff),
                           (unsigned)(a & 0xfff),
                           (unsigned)((b >> 48) & 0xffff) | 0x8000,
                           (unsigned long)(b & 0xffffffffffffUL));
}

static char*
make_body(uint64_t* rng, size_t target)
{
    char* buf = g_malloc(target + 1);
    static const char alpha[] = "abcdefghijklmnopqrstuvwxyz0123456789 ";
    static const int alpha_n = sizeof(alpha) - 1;
    for (size_t i = 0; i < target; i++)
        buf[i] = alpha[xrng(rng) % alpha_n];
    buf[target] = '\0';
    return buf;
}

// ---------------------------------------------------------------------------
// Account dir helpers

static char*
db_path_for(const char* account_jid)
{
    const char* base = getenv("BENCH_DATA_DIR");
    if (!base || !base[0])
        base = "/tmp/cproof-bench-export";
    auto_gchar gchar* jid_dir = ff_jid_to_dir(account_jid);
    auto_gchar gchar* parent = g_strdup_printf("%s/database/%s", base, jid_dir);
    g_mkdir_with_parents(parent, 0755);
    return g_strdup_printf("%s/chatlog.db", parent);
}

// ---------------------------------------------------------------------------
// Schema (lifted from database_sqlite.c so the seeder can populate without
// going through _sqlite_init's full machinery)

static const char* SCHEMA_DDL = "CREATE TABLE IF NOT EXISTS `ChatLogs` ("
                                "`id` INTEGER PRIMARY KEY AUTOINCREMENT, "
                                "`from_jid` TEXT NOT NULL, "
                                "`to_jid` TEXT NOT NULL, "
                                "`from_resource` TEXT, `to_resource` TEXT, "
                                "`message` TEXT, `timestamp` TEXT, `type` TEXT, "
                                "`stanza_id` TEXT, `archive_id` TEXT, "
                                "`encryption` TEXT, `marked_read` INTEGER, "
                                "`replace_id` TEXT, "
                                "`replaces_db_id` INTEGER, `replaced_by_db_id` INTEGER);"
                                "CREATE TABLE IF NOT EXISTS `DbVersion` ("
                                "`dv_id` INTEGER PRIMARY KEY, `version` INTEGER UNIQUE);"
                                "INSERT OR IGNORE INTO `DbVersion` (`version`) VALUES ('2');"
                                "CREATE INDEX IF NOT EXISTS ChatLogs_timestamp_IDX ON `ChatLogs` (`timestamp`);"
                                "CREATE INDEX IF NOT EXISTS ChatLogs_to_from_jid_IDX ON `ChatLogs` (`to_jid`, `from_jid`);"
                                "CREATE TRIGGER IF NOT EXISTS update_corrected_message "
                                "AFTER INSERT ON ChatLogs FOR EACH ROW WHEN NEW.replaces_db_id IS NOT NULL "
                                "BEGIN UPDATE ChatLogs SET replaced_by_db_id = NEW.id "
                                "WHERE id = NEW.replaces_db_id; END;";

// ---------------------------------------------------------------------------
// Seed: open sqlite3 directly, build schema, batched-INSERT N rows.

typedef struct
{
    const char* db_path;
    int64_t rows;
    int contacts;
    int lmc_pct;
    size_t avg_body;
    uint64_t seed;
    const char* account_jid;
} seed_opts_t;

static int
do_seed(const seed_opts_t* o)
{
    sqlite3* db = NULL;
    if (sqlite3_open(o->db_path, &db) != SQLITE_OK) {
        fprintf(stderr, "seed: cannot open %s: %s\n", o->db_path, sqlite3_errmsg(db));
        return 2;
    }
    char* err = NULL;
    if (sqlite3_exec(db, SCHEMA_DDL, NULL, NULL, &err) != SQLITE_OK) {
        fprintf(stderr, "seed: schema failed: %s\n", err);
        sqlite3_free(err);
        sqlite3_close(db);
        return 2;
    }
    sqlite3_exec(db, "PRAGMA journal_mode=WAL;", NULL, NULL, NULL);
    sqlite3_exec(db, "PRAGMA synchronous=NORMAL;", NULL, NULL, NULL);

    if (sqlite3_exec(db, "BEGIN TRANSACTION;", NULL, NULL, &err) != SQLITE_OK) {
        fprintf(stderr, "seed: BEGIN failed: %s\n", err);
        sqlite3_free(err);
        sqlite3_close(db);
        return 2;
    }

    const char* INSERT_SQL = "INSERT INTO ChatLogs ("
                             "from_jid, to_jid, from_resource, to_resource, message, timestamp, type, "
                             "stanza_id, archive_id, encryption, marked_read, replace_id"
                             ") VALUES (?, ?, ?, ?, ?, ?, 'chat', ?, ?, 'none', -1, ?);";
    sqlite3_stmt* stmt = NULL;
    if (sqlite3_prepare_v2(db, INSERT_SQL, -1, &stmt, NULL) != SQLITE_OK) {
        fprintf(stderr, "seed: prepare failed: %s\n", sqlite3_errmsg(db));
        sqlite3_close(db);
        return 2;
    }

    uint64_t rng = o->seed ? o->seed : 1;
    GDateTime* base = g_date_time_new_utc(2020, 1, 1, 0, 0, 0.0);
    int64_t span_secs = (int64_t)5 * 365 * 24 * 3600; // 5 years
    char* prev_sid_per_contact[1024] = { 0 };
    int n_contacts = o->contacts > 0 ? o->contacts : 1;
    if (n_contacts > 1024)
        n_contacts = 1024;

    double t0 = bench_now_ms();

    for (int64_t i = 0; i < o->rows; i++) {
        int ci = (int)(i % n_contacts);
        auto_gchar gchar* contact_jid = g_strdup_printf("buddy%03d@bench.example", ci);

        gboolean is_lmc = prev_sid_per_contact[ci] != NULL && o->lmc_pct > 0
                          && (int)(xrng(&rng) % 100) < o->lmc_pct;

        int64_t off_secs = (int64_t)((double)i / (double)o->rows * (double)span_secs);
        GDateTime* ts = g_date_time_add_seconds(base, off_secs);
        auto_gchar gchar* iso = g_date_time_format_iso8601(ts);
        g_date_time_unref(ts);

        auto_gchar gchar* sid = make_uuid(&rng);
        auto_gchar gchar* aid = (xrng(&rng) & 1) ? make_uuid(&rng) : NULL;
        size_t body_len = o->avg_body > 0 ? o->avg_body : 50 + (xrng(&rng) % 200);
        auto_gchar gchar* body = make_body(&rng, body_len);
        auto_gchar gchar* res = g_strdup_printf("res-%d", (int)(xrng(&rng) % 3));

        sqlite3_bind_text(stmt, 1, contact_jid, -1, SQLITE_TRANSIENT);
        sqlite3_bind_text(stmt, 2, o->account_jid, -1, SQLITE_TRANSIENT);
        sqlite3_bind_text(stmt, 3, res, -1, SQLITE_TRANSIENT);
        sqlite3_bind_null(stmt, 4); // to_resource
        sqlite3_bind_text(stmt, 5, body, -1, SQLITE_TRANSIENT);
        sqlite3_bind_text(stmt, 6, iso, -1, SQLITE_TRANSIENT);
        sqlite3_bind_text(stmt, 7, sid, -1, SQLITE_TRANSIENT);
        if (aid)
            sqlite3_bind_text(stmt, 8, aid, -1, SQLITE_TRANSIENT);
        else
            sqlite3_bind_null(stmt, 8);
        if (is_lmc)
            sqlite3_bind_text(stmt, 9, prev_sid_per_contact[ci], -1, SQLITE_TRANSIENT);
        else
            sqlite3_bind_null(stmt, 9);

        if (sqlite3_step(stmt) != SQLITE_DONE) {
            fprintf(stderr, "seed: insert failed at row %" PRId64 ": %s\n",
                    i, sqlite3_errmsg(db));
            sqlite3_finalize(stmt);
            sqlite3_close(db);
            return 2;
        }
        sqlite3_reset(stmt);

        if (!is_lmc) {
            g_free(prev_sid_per_contact[ci]);
            prev_sid_per_contact[ci] = g_strdup(sid);
        }

        if (((i + 1) % 100000) == 0) {
            double dt = bench_now_ms() - t0;
            fprintf(stderr, "  seeded %" PRId64 " / %" PRId64 " (%.1fs, %.0f rows/s)\n",
                    i + 1, o->rows, dt / 1000.0, (double)(i + 1) / (dt / 1000.0 + 1e-9));
        }
    }

    sqlite3_finalize(stmt);
    if (sqlite3_exec(db, "COMMIT;", NULL, NULL, &err) != SQLITE_OK) {
        fprintf(stderr, "seed: COMMIT failed: %s\n", err);
        sqlite3_free(err);
    }
    sqlite3_close(db);
    g_date_time_unref(base);
    for (int i = 0; i < n_contacts; i++)
        g_free(prev_sid_per_contact[i]);

    double dt = bench_now_ms() - t0;
    fprintf(stderr, "seed: done %" PRId64 " rows in %.2fs\n", o->rows, dt / 1000.0);
    return 0;
}

// ---------------------------------------------------------------------------
// Init dispatcher (sets active_db_backend = sqlite, runs _sqlite_init)

static gboolean
init_sqlite_backend(const char* account_jid)
{
    setenv("BENCH_ACCOUNT_JID", account_jid, 1);
    db_backend_t* be = db_backend_sqlite();
    if (!be) {
        fprintf(stderr, "init: db_backend_sqlite() returned NULL\n");
        return FALSE;
    }
    active_db_backend = be;
    ProfAccount fake = { 0 };
    fake.jid = (gchar*)account_jid;
    if (!be->init(&fake)) {
        fprintf(stderr, "init: backend init failed\n");
        return FALSE;
    }
    return TRUE;
}

static void
close_sqlite_backend(void)
{
    if (active_db_backend && active_db_backend->close)
        active_db_backend->close();
    active_db_backend = NULL;
}

// ---------------------------------------------------------------------------
// SQLite row count

static int64_t
db_row_count(const char* db_path)
{
    sqlite3* db = NULL;
    if (sqlite3_open(db_path, &db) != SQLITE_OK)
        return -1;
    int64_t n = -1;
    sqlite3_stmt* stmt = NULL;
    if (sqlite3_prepare_v2(db, "SELECT COUNT(*) FROM ChatLogs;", -1, &stmt, NULL) == SQLITE_OK) {
        if (sqlite3_step(stmt) == SQLITE_ROW)
            n = sqlite3_column_int64(stmt, 0);
        sqlite3_finalize(stmt);
    }
    sqlite3_close(db);
    return n;
}

// Full content diff: dump every ChatLogs row from both DBs ordered by
// (timestamp, stanza_id), compare tuple-wise. Returns count of mismatches.
// Tuples compared: (from_jid, to_jid, message, timestamp, type, stanza_id,
//                   archive_id, encryption, replace_id).
// marked_read intentionally excluded — it's not faithfully roundtripped
// (export reads -1/null from SQLite, flatfile may emit it differently).
static int64_t
db_diff_full(const char* a_path, const char* b_path, int64_t* total_a, int64_t* total_b)
{
    sqlite3* da = NULL;
    sqlite3* db = NULL;
    if (sqlite3_open(a_path, &da) != SQLITE_OK)
        return -1;
    if (sqlite3_open(b_path, &db) != SQLITE_OK) {
        sqlite3_close(da);
        return -1;
    }

    const char* SQL = "SELECT IFNULL(from_jid,''), IFNULL(to_jid,''), IFNULL(message,''), "
                      "IFNULL(timestamp,''), IFNULL(type,''), IFNULL(stanza_id,''), "
                      "IFNULL(archive_id,''), IFNULL(encryption,''), IFNULL(replace_id,'') "
                      "FROM ChatLogs ORDER BY timestamp ASC, stanza_id ASC;";
    sqlite3_stmt* sa = NULL;
    sqlite3_stmt* sb = NULL;
    if (sqlite3_prepare_v2(da, SQL, -1, &sa, NULL) != SQLITE_OK
        || sqlite3_prepare_v2(db, SQL, -1, &sb, NULL) != SQLITE_OK) {
        if (sa)
            sqlite3_finalize(sa);
        if (sb)
            sqlite3_finalize(sb);
        sqlite3_close(da);
        sqlite3_close(db);
        return -1;
    }

    int64_t na = 0, nb = 0, mismatches = 0;
    int shown = 0;
    while (1) {
        int ra = sqlite3_step(sa);
        int rb = sqlite3_step(sb);
        if (ra != SQLITE_ROW && rb != SQLITE_ROW)
            break;
        if (ra == SQLITE_ROW)
            na++;
        if (rb == SQLITE_ROW)
            nb++;
        if (ra != rb) {
            mismatches++;
            if (shown < 5) {
                fprintf(stderr, "  diff: row count drift at ra_row=%" PRId64 " rb_row=%" PRId64 " ra=%d rb=%d\n", na, nb, ra, rb);
                shown++;
            }
            continue;
        }
        for (int c = 0; c < 9; c++) {
            const unsigned char* va = sqlite3_column_text(sa, c);
            const unsigned char* vb = sqlite3_column_text(sb, c);
            const char* sva = (const char*)(va ? va : (const unsigned char*)"");
            const char* svb = (const char*)(vb ? vb : (const unsigned char*)"");
            if (g_strcmp0(sva, svb) != 0) {
                mismatches++;
                if (shown < 5) {
                    fprintf(stderr, "  diff: row %" PRId64 " col=%d a=\"%s\" b=\"%s\"\n",
                            na, c, sva, svb);
                    shown++;
                }
                break;
            }
        }
    }
    sqlite3_finalize(sa);
    sqlite3_finalize(sb);
    sqlite3_close(da);
    sqlite3_close(db);
    if (total_a)
        *total_a = na;
    if (total_b)
        *total_b = nb;
    return mismatches;
}

// ---------------------------------------------------------------------------
// CLI dispatch

static void
usage(void)
{
    fputs(
        "Usage: bench_export_import SUBCMD [options]\n\n"
        "Subcommands:\n"
        "  seed       --db=PATH --rows=N [--contacts=K] [--lmc=PCT] [--body=BYTES] [--seed=S] [--account=JID]\n"
        "  export     --account=JID --csv=PATH [--label=TAG]\n"
        "  import     --account=JID --csv=PATH [--label=TAG]\n"
        "  roundtrip  --rows=N --csv=PATH [--label=TAG] [--lmc=PCT] [--body=BYTES] [--full-diff]\n"
        "  verify     --db-a=PATH --db-b=PATH\n",
        stderr);
}

static int
cmd_seed(int argc, char** argv)
{
    seed_opts_t o = { 0 };
    o.account_jid = "bench@bench.example";
    o.rows = 1000;
    o.contacts = 1;
    o.lmc_pct = 0;
    o.avg_body = 0;
    o.seed = 42;
    char* db_path = NULL;
    for (int i = 0; i < argc; i++) {
        const char* a = argv[i];
        if (strncmp(a, "--db=", 5) == 0)
            db_path = g_strdup(a + 5);
        else if (strncmp(a, "--rows=", 7) == 0)
            o.rows = strtoll(a + 7, NULL, 10);
        else if (strncmp(a, "--contacts=", 11) == 0)
            o.contacts = atoi(a + 11);
        else if (strncmp(a, "--lmc=", 6) == 0)
            o.lmc_pct = atoi(a + 6);
        else if (strncmp(a, "--body=", 7) == 0)
            o.avg_body = strtoull(a + 7, NULL, 10);
        else if (strncmp(a, "--seed=", 7) == 0)
            o.seed = strtoull(a + 7, NULL, 10);
        else if (strncmp(a, "--account=", 10) == 0)
            o.account_jid = a + 10;
    }
    if (!db_path)
        db_path = db_path_for(o.account_jid);
    o.db_path = db_path;
    int r = do_seed(&o);
    g_free(db_path);
    return r;
}

static int
cmd_export(int argc, char** argv)
{
    const char* account = "bench@bench.example";
    const char* csv = NULL;
    const char* label = "S7_export";
    const char* volume = "export";
    for (int i = 0; i < argc; i++) {
        const char* a = argv[i];
        if (strncmp(a, "--account=", 10) == 0)
            account = a + 10;
        else if (strncmp(a, "--csv=", 6) == 0)
            csv = a + 6;
        else if (strncmp(a, "--label=", 8) == 0)
            label = a + 8;
        else if (strncmp(a, "--volume=", 9) == 0)
            volume = a + 9;
    }
    if (!init_sqlite_backend(account))
        return 2;

    auto_gchar gchar* db_path = db_path_for(account);
    int64_t rows_before = db_row_count(db_path);

    double t0 = bench_now_ms();
    int exported = log_database_export_to_flatfile(NULL);
    double dt = bench_now_ms() - t0;
    long rss = bench_peak_rss_kb();

    fprintf(stderr, "  %s: %d rows exported in %.2fs (db_rows=%" PRId64 ")\n",
            label, exported, dt / 1000.0, rows_before);
    if (csv) {
        struct stat st;
        uint64_t db_size = (stat(db_path, &st) == 0) ? (uint64_t)st.st_size : 0;
        auto_gchar gchar* note = g_strdup_printf(
            "exported=%d db_rows=%" PRId64, exported, rows_before);
        bench_csv_append(csv, label, volume, db_size, (uint64_t)rows_before, dt, rss, note);
    }
    close_sqlite_backend();
    return exported >= 0 ? 0 : 1;
}

static int
cmd_import(int argc, char** argv)
{
    const char* account = "bench@bench.example";
    const char* csv = NULL;
    const char* label = "S8_import";
    const char* volume = "import";
    for (int i = 0; i < argc; i++) {
        const char* a = argv[i];
        if (strncmp(a, "--account=", 10) == 0)
            account = a + 10;
        else if (strncmp(a, "--csv=", 6) == 0)
            csv = a + 6;
        else if (strncmp(a, "--label=", 8) == 0)
            label = a + 8;
        else if (strncmp(a, "--volume=", 9) == 0)
            volume = a + 9;
    }
    if (!init_sqlite_backend(account))
        return 2;

    auto_gchar gchar* db_path = db_path_for(account);
    int64_t rows_before = db_row_count(db_path);

    double t0 = bench_now_ms();
    int imported = log_database_import_from_flatfile(NULL);
    double dt = bench_now_ms() - t0;
    long rss = bench_peak_rss_kb();

    int64_t rows_after = db_row_count(db_path);

    fprintf(stderr, "  %s: %d imported, db rows %" PRId64 " -> %" PRId64 " in %.2fs\n",
            label, imported, rows_before, rows_after, dt / 1000.0);
    if (csv) {
        struct stat st;
        uint64_t db_size = (stat(db_path, &st) == 0) ? (uint64_t)st.st_size : 0;
        auto_gchar gchar* note = g_strdup_printf(
            "imported=%d rows_before=%" PRId64 " rows_after=%" PRId64,
            imported, rows_before, rows_after);
        bench_csv_append(csv, label, volume, db_size,
                         (uint64_t)(rows_after - rows_before), dt, rss, note);
    }
    close_sqlite_backend();
    return imported >= 0 ? 0 : 1;
}

// Roundtrip: seed_A → export → import to fresh DB_B → diff(A, B)
static int
cmd_roundtrip(int argc, char** argv)
{
    int64_t rows = 10000;
    const char* csv = NULL;
    const char* label = "S8e_roundtrip";
    const char* volume = "roundtrip";
    int lmc = 0;
    size_t body = 0;
    int do_diff = 0;
    for (int i = 0; i < argc; i++) {
        const char* a = argv[i];
        if (strncmp(a, "--rows=", 7) == 0)
            rows = strtoll(a + 7, NULL, 10);
        else if (strncmp(a, "--csv=", 6) == 0)
            csv = a + 6;
        else if (strncmp(a, "--label=", 8) == 0)
            label = a + 8;
        else if (strncmp(a, "--volume=", 9) == 0)
            volume = a + 9;
        else if (strncmp(a, "--lmc=", 6) == 0)
            lmc = atoi(a + 6);
        else if (strncmp(a, "--body=", 7) == 0)
            body = strtoull(a + 7, NULL, 10);
        else if (strcmp(a, "--full-diff") == 0)
            do_diff = 1;
    }

    const char* account_a = "rt-a@bench.example";
    const char* account_b = "rt-b@bench.example";
    auto_gchar gchar* db_a = db_path_for(account_a);
    auto_gchar gchar* db_b = db_path_for(account_b);
    unlink(db_a);
    unlink(db_b);
    auto_gchar gchar* wal_a = g_strdup_printf("%s-wal", db_a);
    unlink(wal_a);
    auto_gchar gchar* shm_a = g_strdup_printf("%s-shm", db_a);
    unlink(shm_a);
    auto_gchar gchar* wal_b = g_strdup_printf("%s-wal", db_b);
    unlink(wal_b);
    auto_gchar gchar* shm_b = g_strdup_printf("%s-shm", db_b);
    unlink(shm_b);

    fprintf(stderr, "===== roundtrip: rows=%" PRId64 " account_a=%s account_b=%s =====\n",
            rows, account_a, account_b);

    // 1) Seed DB_A
    seed_opts_t so = { 0 };
    so.db_path = db_a;
    so.rows = rows;
    so.contacts = 1;
    so.lmc_pct = lmc;
    so.avg_body = body;
    so.seed = 42;
    so.account_jid = account_a;
    double seed_ms_t0 = bench_now_ms();
    if (do_seed(&so) != 0)
        return 2;
    double seed_ms = bench_now_ms() - seed_ms_t0;

    // 2) Export DB_A → flatlog under account_a
    if (!init_sqlite_backend(account_a))
        return 2;
    double export_t0 = bench_now_ms();
    int exported = log_database_export_to_flatfile(NULL);
    double export_ms = bench_now_ms() - export_t0;
    close_sqlite_backend();
    fprintf(stderr, "  exported %d rows in %.2fs\n", exported, export_ms / 1000.0);

    // 3) Cross-mount the flatlog tree at account_b's expected location
    const char* base = getenv("BENCH_DATA_DIR");
    if (!base || !base[0])
        base = "/tmp/cproof-bench-export";
    auto_gchar gchar* dir_a = g_strdup_printf("%s/flatlog/%s", base, ff_jid_to_dir(account_a));
    auto_gchar gchar* dir_b = g_strdup_printf("%s/flatlog/%s", base, ff_jid_to_dir(account_b));
    auto_gchar gchar* parent_b = g_path_get_dirname(dir_b);
    g_mkdir_with_parents(parent_b, 0755);
    // Symlink dir_a as dir_b so import on account_b reads the same files.
    unlink(dir_b);
    if (symlink(dir_a, dir_b) != 0 && errno != EEXIST) {
        fprintf(stderr, "  symlink %s -> %s failed: %s\n", dir_b, dir_a, strerror(errno));
        return 2;
    }

    // 4) Import flatlog → DB_B
    if (!init_sqlite_backend(account_b))
        return 2;
    double import_t0 = bench_now_ms();
    int imported = log_database_import_from_flatfile(NULL);
    double import_ms = bench_now_ms() - import_t0;
    close_sqlite_backend();
    fprintf(stderr, "  imported %d rows in %.2fs\n", imported, import_ms / 1000.0);

    int64_t rows_a = db_row_count(db_a);
    int64_t rows_b = db_row_count(db_b);
    fprintf(stderr, "  row counts: A=%" PRId64 " B=%" PRId64 "\n", rows_a, rows_b);

    int64_t mismatches = 0;
    double diff_ms = 0;
    if (do_diff) {
        double diff_t0 = bench_now_ms();
        int64_t na = 0, nb = 0;
        mismatches = db_diff_full(db_a, db_b, &na, &nb);
        diff_ms = bench_now_ms() - diff_t0;
        fprintf(stderr, "  full content diff: a_rows=%" PRId64 " b_rows=%" PRId64 " mismatches=%" PRId64 " (%.2fs)\n",
                na, nb, mismatches, diff_ms / 1000.0);
    }

    long rss = bench_peak_rss_kb();
    double total_ms = seed_ms + export_ms + import_ms + diff_ms;

    if (csv) {
        struct stat st;
        uint64_t db_size = (stat(db_a, &st) == 0) ? (uint64_t)st.st_size : 0;
        auto_gchar gchar* note = g_strdup_printf(
            "rows=%" PRId64 " seed=%.0fms export=%.0fms import=%.0fms diff=%.0fms "
            "exported=%d imported=%d rows_a=%" PRId64 " rows_b=%" PRId64
            " mismatches=%" PRId64 " body=%zu lmc=%d",
            rows, seed_ms, export_ms, import_ms, diff_ms,
            exported, imported, rows_a, rows_b, mismatches, body, lmc);
        bench_csv_append(csv, label, volume, db_size, (uint64_t)rows,
                         total_ms, rss, note);
    }

    int ok = (rows_a == rows_b)
             && (!do_diff || mismatches == 0)
             && exported >= 0 && imported >= 0;
    fprintf(stderr, "  %s\n", ok ? "PASS" : "FAIL");
    return ok ? 0 : 1;
}

static int
cmd_verify(int argc, char** argv)
{
    const char* a = NULL;
    const char* b = NULL;
    for (int i = 0; i < argc; i++) {
        const char* x = argv[i];
        if (strncmp(x, "--db-a=", 7) == 0)
            a = x + 7;
        else if (strncmp(x, "--db-b=", 7) == 0)
            b = x + 7;
    }
    if (!a || !b) {
        usage();
        return 2;
    }
    int64_t na = 0, nb = 0;
    int64_t m = db_diff_full(a, b, &na, &nb);
    fprintf(stderr, "verify: a_rows=%" PRId64 " b_rows=%" PRId64 " mismatches=%" PRId64 "\n",
            na, nb, m);
    return (m == 0 && na == nb) ? 0 : 1;
}

int
main(int argc, char** argv)
{
    if (argc < 2) {
        usage();
        return 2;
    }
    const char* sub = argv[1];
    int sub_argc = argc - 2;
    char** sub_argv = argv + 2;
    if (g_strcmp0(sub, "seed") == 0)
        return cmd_seed(sub_argc, sub_argv);
    if (g_strcmp0(sub, "export") == 0)
        return cmd_export(sub_argc, sub_argv);
    if (g_strcmp0(sub, "import") == 0)
        return cmd_import(sub_argc, sub_argv);
    if (g_strcmp0(sub, "roundtrip") == 0)
        return cmd_roundtrip(sub_argc, sub_argv);
    if (g_strcmp0(sub, "verify") == 0)
        return cmd_verify(sub_argc, sub_argv);
    usage();
    return 2;
}