inline int getLastAuthType() const { return last_auth_type_; }

    int last_auth_type_ = 0;

#include <unistd.h>
#include <string>
#include <sstream>
#include <mysql.h>          // admin interface is reached via the MySQL client
#include "pg_lite_client.h"
#include "command_line.h"
#include "tap.h"
#include "utils.h"

CommandLine cl;

// Admin connection (MySQL protocol) used to flip pgsql-authentication_method.
static MYSQL* admin_connect() {
    MYSQL* conn = mysql_init(NULL);
    if (!mysql_real_connect(conn, cl.admin_host, cl.admin_username, cl.admin_password,
                            NULL, cl.admin_port, NULL, 0)) {
        diag("admin connect failed: %s", mysql_error(conn));
        mysql_close(conn);
        return NULL;
    }
    return conn;
}

static bool set_frontend_auth_method(MYSQL* admin, int method) {
    std::string q = "SET pgsql-authentication_method=" + std::to_string(method);
    if (mysql_query(admin, q.c_str())) { diag("SET failed: %s", mysql_error(admin)); return false; }
    if (mysql_query(admin, "LOAD PGSQL VARIABLES TO RUNTIME")) { diag("LOAD failed: %s", mysql_error(admin)); return false; }
    return true;
}

// Attempts a frontend login with pg_lite_client, running a query to prove the
// session is usable. On success, observed_auth_type = the challenge type ProxySQL
// presented (3=cleartext, 5=md5, 10=scram). Returns true on successful auth+query.
static bool try_frontend_login(const std::string& user, const std::string& password,
                               int& observed_auth_type) {
    observed_auth_type = 0;
    try {
        PgConnection c(2000);
        c.connect(cl.pgsql_host, cl.pgsql_port, user /*dbname==user in this infra*/, user, password);
        observed_auth_type = c.getLastAuthType();
        c.execute("SELECT 1");   // #5865 runs NO queries; proving the session works is our value-add
        c.disconnect();
        return true;
    } catch (const PgException& e) {
        diag("login threw: %s", e.what());
        return false;
    }
}

int main(int argc, char** argv) {
    if (cl.getEnv()) return exit_status();

    // Per method: (login succeeds + query runs) AND (observed challenge type matches floor).
    // Task 1 lands cleartext only (2 assertions); Tasks 2-3 add md5, scram, and failures.
    plan(2);

    MYSQL* admin = admin_connect();
    if (!admin) BAIL_OUT("cannot reach admin");

    // --- Cleartext floor (method = 1) -> expect challenge type 3 on the wire ---
    set_frontend_auth_method(admin, 1);   // affects NEW frontend connections
    int auth_type = 0;
    bool logged_in = try_frontend_login(cl.pgsql_username, cl.pgsql_password, auth_type);
    ok(logged_in, "cleartext floor: login + query succeed");
    ok(auth_type == 3, "cleartext floor: ProxySQL presented challenge type 3 (got %d)", auth_type);

    // restore default before exit
    set_frontend_auth_method(admin, 3);
    mysql_close(admin);
    return exit_status();
}

pgsql-auth_method_matrix-t: pgsql-auth_method_matrix-t.cpp pg_lite_client.cpp $(TAP_LDIR)/libtap.so
	$(CXX) $< pg_lite_client.cpp $(IDIRS) $(LDIRS) $(OPT) $(MYLIBS) $(STATIC_LIBS) -lscram -lusual -Wl,--allow-multiple-definition -o $@

  "pgsql-auth_method_matrix-t" : [ "legacy-g4","mysql-auto_increment_delay_multiplex=0-g4","mysql-multiplexing=false-g4","mysql-query_digests=0-g4","mysql-query_digests_keep_comment=1-g4" ],

PROXYSQL31=1 make -j$(nproc) debug
PROXYSQL31=1 make build_tap_test_debug
WORKSPACE=$(pwd) INFRA_ID=dev-$USER TAP_GROUP=legacy-g4 test/infra/control/ensure-infras.bash

WORKSPACE=$(pwd) INFRA_ID=dev-$USER TAP_GROUP=legacy-g4 \
  TEST_PY_TAP_INCL="pgsql-auth_method_matrix-t" \
  test/infra/control/run-tests-isolated.bash

git add test/tap/tests/pgsql-auth_method_matrix-t.cpp test/tap/tests/Makefile test/tap/groups/groups.json
git commit -m "test(pgsql): auth-method matrix scaffold + cleartext case"

    // --- MD5 floor (method = 2) -> expect challenge type 5 on the wire ---
    set_frontend_auth_method(admin, 2);
    int md5_auth = 0;
    ok(try_frontend_login(cl.pgsql_username, cl.pgsql_password, md5_auth),
       "md5 floor: login + query succeed");
    ok(md5_auth == 5, "md5 floor: ProxySQL presented challenge type 5 (got %d)", md5_auth);

PROXYSQL31=1 make build_tap_test_debug
WORKSPACE=$(pwd) INFRA_ID=dev-$USER TAP_GROUP=legacy-g4 \
  TEST_PY_TAP_INCL="pgsql-auth_method_matrix-t" test/infra/control/run-tests-isolated.bash

    void sendMD5Password(const std::string& password, const uint8_t salt[4]);

#include <openssl/md5.h>

static std::string md5_hex(const std::string& in) {
    unsigned char digest[MD5_DIGEST_LENGTH];
    MD5(reinterpret_cast<const unsigned char*>(in.data()), in.size(), digest);
    static const char* hx = "0123456789abcdef";
    std::string out;
    out.reserve(MD5_DIGEST_LENGTH * 2);
    for (int i = 0; i < MD5_DIGEST_LENGTH; ++i) {
        out.push_back(hx[digest[i] >> 4]);
        out.push_back(hx[digest[i] & 0x0f]);
    }
    return out;
}

// PostgreSQL MD5 auth: "md5" + md5( md5(password + user) + salt )
void PgConnection::sendMD5Password(const std::string& password, const uint8_t salt[4]) {
    std::string inner = md5_hex(password + user_);
    std::string with_salt = inner;
    with_salt.append(reinterpret_cast<const char*>(salt), 4);
    std::string token = "md5" + md5_hex(with_salt);
    std::vector<uint8_t> packet;
    writeStringToBuffer(packet, token);   // null-terminated C string
    sendMessage('p', packet);
}

            else if (authType == 5) {  // AuthenticationMD5Password (4-byte salt follows)
                if (buffer.size() < 8) throw PgException("Invalid MD5 auth message");
                uint8_t salt[4];
                memcpy(salt, buffer.data() + 4, 4);
                sendMD5Password(password, salt);
                readMessage(type, buffer);
                if (type == AUTH_TYPE) {
                    authType = ntohl(*reinterpret_cast<int32_t*>(buffer.data()));
                    if (authType == 0) return;
                }
                // fall through to error handling on non-OK
            }

PROXYSQL31=1 make build_tap_test_debug
WORKSPACE=$(pwd) INFRA_ID=dev-$USER TAP_GROUP=legacy-g4 \
  TEST_PY_TAP_INCL="pgsql-auth_method_matrix-t" test/infra/control/run-tests-isolated.bash

git add test/tap/tests/pg_lite_client.h test/tap/tests/pg_lite_client.cpp test/tap/tests/pgsql-auth_method_matrix-t.cpp
git commit -m "test(pgsql): pg_lite_client MD5 frontend auth + matrix md5 case"

    // --- SCRAM floor (method = 3) -> expect challenge type 10 on the wire ---
    set_frontend_auth_method(admin, 3);
    int scram_auth = 0;
    ok(try_frontend_login(cl.pgsql_username, cl.pgsql_password, scram_auth),
       "scram floor: login + query succeed");
    ok(scram_auth == 10, "scram floor: ProxySQL presented SASL/SCRAM challenge type 10 (got %d)", scram_auth);

    // --- Wrong-password failure paths, one per floor (challenge type irrelevant) ---
    int ignore = 0;
    set_frontend_auth_method(admin, 1);
    ok(!try_frontend_login(cl.pgsql_username, "wrong-pw", ignore), "cleartext floor: wrong password rejected");
    set_frontend_auth_method(admin, 2);
    ok(!try_frontend_login(cl.pgsql_username, "wrong-pw", ignore), "md5 floor: wrong password rejected");
    set_frontend_auth_method(admin, 3);
    ok(!try_frontend_login(cl.pgsql_username, "wrong-pw", ignore), "scram floor: wrong password rejected");

PROXYSQL31=1 make build_tap_test_debug
WORKSPACE=$(pwd) INFRA_ID=dev-$USER TAP_GROUP=legacy-g4 \
  TEST_PY_TAP_INCL="pgsql-auth_method_matrix-t" test/infra/control/run-tests-isolated.bash

    void doSASLAuth(const std::string& password, const std::vector<uint8_t>& mechListMsg);

extern "C" {
#include "scram.h"
}

// Completes a SCRAM-SHA-256 SASL exchange as the CLIENT, reusing deps/libscram.
// mechListMsg is the AuthenticationSASL(10) payload after the 4-byte authType:
// a sequence of null-terminated mechanism names terminated by an extra null.
void PgConnection::doSASLAuth(const std::string& password,
                              const std::vector<uint8_t>& /*mechListMsg*/) {
    ScramState* st = scram_state_init();
    PgCredentials cred;
    memset(&cred, 0, sizeof(cred));
    snprintf(cred.name, sizeof(cred.name), "%s", user_.c_str());
    snprintf(cred.passwd, sizeof(cred.passwd), "%s", password.c_str());
    cred.has_scram_keys = false;

    // 1) SASLInitialResponse ('p'): mechanism name + Int32 length + client-first-message
    char* client_first = build_client_first_message(st);   // e.g. "n,,n=,r=<nonce>"
    if (!client_first) { free_scram_state(st); throw PgException(std::string("scram client-first: ") + scram_error()); }
    {
        std::vector<uint8_t> pkt;
        const char* mech = "SCRAM-SHA-256";
        writeStringToBuffer(pkt, mech);                    // null-terminated mechanism
        int32_t clen = htonl((int32_t)strlen(client_first));
        const uint8_t* cp = reinterpret_cast<const uint8_t*>(&clen);
        pkt.insert(pkt.end(), cp, cp + 4);                 // Int32 length
        pkt.insert(pkt.end(), client_first, client_first + strlen(client_first));
        sendMessage('p', pkt);
    }

    // 2) Expect AuthenticationSASLContinue (authType 11) with server-first-message
    char type; std::vector<uint8_t> buffer;
    readMessage(type, buffer);
    if (type != AUTH_TYPE || buffer.size() < 4 ||
        ntohl(*reinterpret_cast<int32_t*>(buffer.data())) != 11) {
        free(client_first); free_scram_state(st);
        throw PgException("expected AuthenticationSASLContinue(11)");
    }
    std::string server_first(reinterpret_cast<const char*>(buffer.data()) + 4, buffer.size() - 4);
    char* server_nonce = nullptr; char* salt = nullptr; int saltlen = 0; int iterations = 0;
    if (!read_server_first_message(st, const_cast<char*>(server_first.c_str()),
                                   &server_nonce, &salt, &saltlen, &iterations)) {
        free(client_first); free_scram_state(st);
        throw PgException(std::string("scram read server-first: ") + scram_error());
    }

    // 3) SASLResponse ('p'): client-final-message (with proof)
    char* client_final = build_client_final_message(st, &cred, server_nonce, salt, saltlen, iterations);
    if (!client_final) { free(client_first); free_scram_state(st); throw PgException(std::string("scram client-final: ") + scram_error()); }
    {
        std::vector<uint8_t> pkt(client_final, client_final + strlen(client_final));
        sendMessage('p', pkt);
    }

    // 4) Expect AuthenticationSASLFinal (authType 12) with server-final (v=ServerSignature)
    readMessage(type, buffer);
    if (type != AUTH_TYPE || buffer.size() < 4 ||
        ntohl(*reinterpret_cast<int32_t*>(buffer.data())) != 12) {
        free(client_first); free(client_final); free_scram_state(st);
        throw PgException("expected AuthenticationSASLFinal(12)");
    }
    {
        std::string server_final(reinterpret_cast<const char*>(buffer.data()) + 4, buffer.size() - 4);
        char server_sig[256] = {0};
        if (!read_server_final_message(const_cast<char*>(server_final.c_str()), server_sig) ||
            !verify_server_signature(st, &cred, server_sig)) {
            free(client_first); free(client_final); free_scram_state(st);
            throw PgException("scram server signature verification failed");
        }
    }
    free(client_first); free(client_final); free_scram_state(st);

    // 5) Expect AuthenticationOk (0)
    readMessage(type, buffer);
    if (type == AUTH_TYPE && ntohl(*reinterpret_cast<int32_t*>(buffer.data())) == 0) return;
    throw PgException("scram: no AuthenticationOk after SASLFinal");
}

            else if (authType == 10) {  // AuthenticationSASL (mechanism list follows)
                doSASLAuth(password, buffer);
                return;   // doSASLAuth consumes through AuthenticationOk
            }

PROXYSQL31=1 make build_tap_test_debug
WORKSPACE=$(pwd) INFRA_ID=dev-$USER TAP_GROUP=legacy-g4 \
  TEST_PY_TAP_INCL="pgsql-auth_method_matrix-t" test/infra/control/run-tests-isolated.bash

git add test/tap/tests/pg_lite_client.h test/tap/tests/pg_lite_client.cpp test/tap/tests/pgsql-auth_method_matrix-t.cpp
git commit -m "test(pgsql): pg_lite_client SCRAM-SHA-256 frontend auth + matrix scram/failure cases"

    // Requires #5865: create a user whose stored secret is a SCRAM verifier.
    // Reuse #5865's runtime pattern: PQencryptPasswordConn(conn, "verifier_pw", "scram_user", "scram-sha-256")
    // then INSERT INTO pgsql_users(username,password,...) VALUES('scram_user', <that verifier>, ...); LOAD PGSQL USERS TO RUNTIME.
    // (Factor #5865's helper out or copy its ~5-line create_verifier_user() here.)
    create_verifier_user(admin, "scram_user", "verifier_pw");   // from #5865

    set_frontend_auth_method(admin, 1);   // cleartext FLOOR...
    int vt = 0;
    bool v_ok = try_frontend_login("scram_user", "verifier_pw", vt);
    ok(v_ok, "verifier user authenticates under cleartext floor");
    ok(vt == 10, "no-downgrade: verifier user still challenged with SCRAM(10) under cleartext floor (got %d)", vt);

#include <string>
#include <vector>
#include "pg_lite_client.h"
#include "command_line.h"
#include "tap.h"
#include "utils.h"

CommandLine cl;

struct Case {
    const char* label;
    const char* select_expr;   // e.g. "SELECT '\\xdeadbeef'::bytea"
    const char* expected_text; // expected value in TEXT format
    int32_t expected_oid;      // PostgreSQL type OID
    const char* expected_binary_hex; // exact wire bytes in BINARY format
};

// Everything observed for one round-trip of a case at a given result format.
struct Observed {
    bool got_result = false;        // RowDescription+DataRow returned (no ErrorResponse)
    int32_t oid = 0;                // type OID from RowDescription
    int16_t col_format = -1;        // per-column result format code (0=text, 1=binary)
    bool is_null = true;            // DataRow value length == -1 ?
    std::string text_value;         // value decoded as text (meaningful when col_format==0)
    std::vector<uint8_t> raw_bytes; // raw DataRow payload for column 0 (both formats)
};

// Lowercase hex of a raw payload, for comparison and diagnostics.
static std::string to_hex(const std::vector<uint8_t>& bytes);

// One representative literal per type; expand freely -- adding a row is the unit
// of work. expected_binary_hex is the exact payload PostgreSQL's *_send() emits,
// so the binary assertion compares bytes rather than merely checking the OID.
//
// Excerpt only -- the shipped test carries the full table (bool, int4, int8,
// float8, numeric, text_utf8, bytea, uuid, timestamptz, jsonb, int4_array, inet)
// and is the source of truth for the expected encodings:
//   test/tap/tests/pgsql-datatype_matrix-t.cpp
static const std::vector<Case> cases = {
    { "int4",        "SELECT 2147483647::int4",            "2147483647",     23,
      "7fffffff" },
    // A genuine timestamptz (OID 1184). Applying AT TIME ZONE 'UTC' would yield
    // timestamp *without* time zone (OID 1114) and never exercise timestamptz at
    // all; instead the session pins TimeZone=UTC so the text form is deterministic.
    { "timestamptz", "SELECT '2020-01-01 00:00:00+00'::timestamptz",
                                                           "2020-01-01 00:00:00+00", 1184,
      "00023e0786c26000" },
    { "bytea",       "SELECT '\\xdeadbeef'::bytea",        "\\xdeadbeef",    17,
      "deadbeef" },
};

// Runs one case through pg_lite_client at the given result format (0=text,1=binary),
// capturing the RowDescription OID + per-column format code and the raw DataRow bytes.
static bool run_case(const Case& c, int16_t fmt, Observed& obs);

int main(int argc, char** argv) {
    if (cl.getEnv()) return exit_status();

    // For each case: 1 text assertion (value+oid) + 1 binary assertion
    // (oid + format code + exact payload bytes).
    plan((int)cases.size() * 2);

    for (const auto& c : cases) {
        Observed t;
        bool ok_text = run_case(c, 0, t);
        ok(ok_text && t.col_format == 0 && t.oid == c.expected_oid && t.text_value == c.expected_text,
           "%s text: oid=%d value='%s'", c.label, t.oid, t.text_value.c_str());

        // The binary half must compare the DataRow payload byte-for-byte against
        // the expected encoding. Asserting only the OID is NOT enough: a silent
        // text fallback, a truncated payload, or a byte-order regression all keep
        // the OID intact and would pass. Each Case therefore carries the exact
        // bytes PostgreSQL's *_send() emits for its literal.
        Observed b;
        bool ok_bin = run_case(c, 1, b);
        ok(ok_bin && b.col_format == 1 && b.oid == c.expected_oid
               && !b.is_null && to_hex(b.raw_bytes) == c.expected_binary_hex,
           "%s binary: oid=%d payload=%s (expected %s)",
           c.label, b.oid, to_hex(b.raw_bytes).c_str(), c.expected_binary_hex);
    }
    return exit_status();
}

static bool run_case(const Case& c, int16_t fmt, Observed& obs) {
    try {
        PgConnection conn(2000);
        conn.connect(cl.pgsql_host, cl.pgsql_port, cl.pgsql_username, cl.pgsql_username, cl.pgsql_password);

        // Pin the session time zone so the TEXT rendering of timestamptz is
        // deterministic regardless of the backend's configured TimeZone.
        conn.execute("SET TIME ZONE 'UTC'");
        conn.consumeInputUntilReady();

        // Extended protocol: unnamed prepared statement, single result format = fmt.
        // These queries take NO bind parameters, so the param-format array must be
        // empty: bindStatementSingleFormat() would send a 1-element param-format
        // array for a 0-param Bind, tripping issue #5899. bindStatementEx() with an
        // explicit empty paramFormats array is the protocol-correct 0-param bind.
        conn.prepareStatement("", c.select_expr, false, {});
        conn.bindStatementEx("", "", {}, {}, { fmt }, false);
        conn.describePortal("", false);
        conn.executePortal("", 0, true);   // sync

        // Read: ParseComplete(1), BindComplete(2), RowDescription(T), DataRow(D), CommandComplete(C), ReadyForQuery(Z)
        char type; std::vector<uint8_t> buf;
        while (true) {
            conn.readMessage(type, buf);
            if (type == PgConnection::ROW_DESCRIPTION) {
                BufferReader r(buf);
                int16_t nfields = r.readInt16();
                if (nfields >= 1) {
                    r.readString();                 // field name
                    r.readInt32();                  // table oid
                    r.readInt16();                  // column attr
                    obs.oid = r.readInt32();        // type oid
                    r.readInt16();                  // type size
                    r.readInt32();                  // type modifier
                    obs.col_format = r.readInt16(); // per-column result FORMAT CODE
                }
            } else if (type == PgConnection::DATA_ROW) {
                BufferReader r(buf);
                int16_t ncols = r.readInt16();
                if (ncols >= 1) {
                    int32_t len = r.readInt32();
                    if (len >= 0) {
                        obs.raw_bytes = r.readBytes(len);
                        obs.is_null = false;
                        // Decoded as text for the text-format assertion; in binary
                        // format the payload is opaque and text_value is unused.
                        obs.text_value.assign(obs.raw_bytes.begin(), obs.raw_bytes.end());
                        obs.got_result = true;
                    } else {
                        obs.is_null = true;
                        obs.got_result = true;
                    }
                }
            } else if (type == PgConnection::READY_FOR_QUERY) {
                break;
            } else if (type == PgConnection::ERROR_RESPONSE) {
                conn.disconnect();
                return false;
            }
        }
        conn.disconnect();
        return obs.got_result;
    } catch (const PgException& e) {
        diag("%s fmt=%d threw: %s", c.label, (int)fmt, e.what());
        return false;
    }
}

pgsql-datatype_matrix-t: pgsql-datatype_matrix-t.cpp pg_lite_client.cpp $(TAP_LDIR)/libtap.so
	$(CXX) $< pg_lite_client.cpp $(IDIRS) $(LDIRS) $(OPT) $(MYLIBS) $(STATIC_LIBS) -o $@

  "pgsql-datatype_matrix-t" : [ "legacy-g4","mysql-auto_increment_delay_multiplex=0-g4","mysql-multiplexing=false-g4","mysql-query_digests=0-g4","mysql-query_digests_keep_comment=1-g4" ],

PROXYSQL31=1 make build_tap_test_debug
WORKSPACE=$(pwd) INFRA_ID=dev-$USER TAP_GROUP=legacy-g4 \
  TEST_PY_TAP_INCL="pgsql-datatype_matrix-t" test/infra/control/run-tests-isolated.bash

git add test/tap/tests/pgsql-datatype_matrix-t.cpp test/tap/tests/Makefile test/tap/groups/groups.json
git commit -m "test(pgsql): data-type/binary-encoding matrix (text+binary, oid+value)"

#include <string>
#include <sstream>
#include "libpq-fe.h"
#include "pg_lite_client.h"
#include "command_line.h"
#include "tap.h"
#include "utils.h"

CommandLine cl;
using PGConnPtr = std::unique_ptr<PGconn, decltype(&PQfinish)>;

static PGConnPtr backend_conn() {
    std::stringstream ss;
    ss << "host=" << cl.pgsql_host << " port=" << cl.pgsql_port
       << " user=" << cl.pgsql_username << " password=" << cl.pgsql_password
       << " dbname=" << cl.pgsql_username << " sslmode=disable";
    PGconn* c = PQconnectdb(ss.str().c_str());
    return PGConnPtr(c, &PQfinish);
}

// Extended-protocol portal suspension: Execute with maxRows=2 over a 5-row result.
static bool portal_suspends_at_2() {
    try {
        PgConnection conn(2000);
        conn.connect(cl.pgsql_host, cl.pgsql_port, cl.pgsql_username, cl.pgsql_username, cl.pgsql_password);
        conn.prepareStatement("", "SELECT g FROM generate_series(1,5) g", false, {});
        conn.bindStatementSingleFormat("", "", {}, 0, { 0 }, false);
        conn.executePortal("", 2, true);   // maxRows=2 -> expect 2 DataRows then PortalSuspended
        char type; std::vector<uint8_t> buf; int rows = 0; bool suspended = false;
        while (true) {
            conn.readMessage(type, buf);
            if (type == PgConnection::DATA_ROW) rows++;
            else if (type == PgConnection::PORTAL_SUSPENDED) { suspended = true; }
            else if (type == PgConnection::READY_FOR_QUERY) break;
            else if (type == PgConnection::ERROR_RESPONSE) { conn.disconnect(); return false; }
        }
        conn.disconnect();
        return rows == 2 && suspended;
    } catch (const PgException& e) { diag("portal test threw: %s", e.what()); return false; }
}

int main(int argc, char** argv) {
    if (cl.getEnv()) return exit_status();
    plan(4);

    PGConnPtr c = backend_conn();
    ok(c && PQstatus(c.get()) == CONNECTION_OK, "connected for cursor test");

    // DECLARE / FETCH / MOVE / CLOSE inside a transaction (cursors require a txn).
    PQexec(c.get(), "BEGIN");
    PQexec(c.get(), "DECLARE cur CURSOR FOR SELECT g FROM generate_series(1,10) g");
    PGresult* r = PQexec(c.get(), "FETCH 3 cur");
    ok(PQntuples(r) == 3, "FETCH 3 returns 3 rows");
    PQclear(r);
    r = PQexec(c.get(), "MOVE 2 cur");           // skip 2
    PQclear(r);
    r = PQexec(c.get(), "FETCH 10 cur");         // remaining 5
    ok(PQntuples(r) == 5, "MOVE 2 then FETCH returns remaining 5 rows");
    PQclear(r);
    PQexec(c.get(), "CLOSE cur");
    PQexec(c.get(), "COMMIT");

    ok(portal_suspends_at_2(), "extended-protocol portal suspends at maxRows=2");

    return exit_status();
}

pgsql-server_side_cursors-t: pgsql-server_side_cursors-t.cpp pg_lite_client.cpp $(TAP_LDIR)/libtap.so
	$(CXX) $< pg_lite_client.cpp $(IDIRS) $(LDIRS) $(OPT) $(MYLIBS) $(STATIC_LIBS) -o $@

  "pgsql-server_side_cursors-t" : [ "legacy-g4","mysql-auto_increment_delay_multiplex=0-g4","mysql-multiplexing=false-g4","mysql-query_digests=0-g4","mysql-query_digests_keep_comment=1-g4" ],

PROXYSQL31=1 make build_tap_test_debug
WORKSPACE=$(pwd) INFRA_ID=dev-$USER TAP_GROUP=legacy-g4 \
  TEST_PY_TAP_INCL="pgsql-server_side_cursors-t" test/infra/control/run-tests-isolated.bash

git add test/tap/tests/pgsql-server_side_cursors-t.cpp test/tap/tests/Makefile test/tap/groups/groups.json
git commit -m "test(pgsql): server-side cursors (DECLARE/FETCH/MOVE) + portal suspension"

#include <string>
#include <sstream>
#include <vector>
#include <memory>
#include "libpq-fe.h"
#include "command_line.h"
#include "tap.h"
#include "utils.h"

CommandLine cl;
using PGConnPtr = std::unique_ptr<PGconn, decltype(&PQfinish)>;

static PGConnPtr mk() {
    std::stringstream ss;
    ss << "host=" << cl.pgsql_host << " port=" << cl.pgsql_port
       << " user=" << cl.pgsql_username << " password=" << cl.pgsql_password
       << " dbname=" << cl.pgsql_username << " sslmode=disable";
    return PGConnPtr(PQconnectdb(ss.str().c_str()), &PQfinish);
}

static std::string scalar(PGconn* c, const char* q) {
    PGresult* r = PQexec(c, q);
    std::string v = (PQresultStatus(r) == PGRES_TUPLES_OK && PQntuples(r) > 0) ? PQgetvalue(r, 0, 0) : "";
    PQclear(r);
    return v;
}

int main(int argc, char** argv) {
    if (cl.getEnv()) return exit_status();
    plan(3);

    // 1) Connection storm: open many short-lived connections; none should error out.
    bool all_ok = true;
    for (int i = 0; i < 100; ++i) {
        PGConnPtr c = mk();
        if (!c || PQstatus(c.get()) != CONNECTION_OK) { all_ok = false; break; }
        if (scalar(c.get(), "SELECT 1") != "1") { all_ok = false; break; }
    }
    ok(all_ok, "100 sequential short connections all succeed (no pool leak/exhaustion)");

    // 2) Session-state isolation across multiplexed reuse.
    // Connection A sets a session GUC; a fresh connection B must NOT observe it.
    {
        PGConnPtr a = mk();
        PQexec(a.get(), "SET application_name = 'churn_A'");
        std::string a_val = scalar(a.get(), "SHOW application_name");
        ok(a_val == "churn_A", "connection A sees its own SET application_name");

        PGConnPtr b = mk();
        std::string b_val = scalar(b.get(), "SHOW application_name");
        ok(b_val != "churn_A", "connection B does NOT inherit A's session state (got '%s')", b_val.c_str());
    }

    return exit_status();
}

  "pgsql-pool_churn-t" : [ "legacy-g4","mysql-auto_increment_delay_multiplex=0-g4","mysql-multiplexing=false-g4","mysql-query_digests=0-g4","mysql-query_digests_keep_comment=1-g4" ],

PROXYSQL31=1 make build_tap_test_debug
WORKSPACE=$(pwd) INFRA_ID=dev-$USER TAP_GROUP=legacy-g4 \
  TEST_PY_TAP_INCL="pgsql-pool_churn-t" test/infra/control/run-tests-isolated.bash

git add test/tap/tests/pgsql-pool_churn-t.cpp test/tap/groups/groups.json
git commit -m "test(pgsql): pool churn + session-state isolation across reuse"

#include <string>
#include <sstream>
#include <mysql.h>
#include "libpq-fe.h"
#include "command_line.h"
#include "tap.h"
#include "utils.h"

CommandLine cl;
using PGConnPtr = std::unique_ptr<PGconn, decltype(&PQfinish)>;

static PGConnPtr mk() {
    std::stringstream ss;
    ss << "host=" << cl.pgsql_host << " port=" << cl.pgsql_port
       << " user=" << cl.pgsql_username << " password=" << cl.pgsql_password
       << " dbname=" << cl.pgsql_username << " sslmode=disable";
    return PGConnPtr(PQconnectdb(ss.str().c_str()), &PQfinish);
}

static std::string sqlstate_of(PGresult* r) {
    const char* s = PQresultErrorField(r, PG_DIAG_SQLSTATE);
    return s ? s : "";
}

int main(int argc, char** argv) {
    if (cl.getEnv()) return exit_status();
    plan(4);

    PGConnPtr c = mk();
    ok(c && PQstatus(c.get()) == CONNECTION_OK, "connected for listen/notify contract");

    // LISTEN over simple protocol -> 0A000 feature_not_supported (libpq path).
    PGresult* r = PQexec(c.get(), "LISTEN chan1");
    ok(PQresultStatus(r) == PGRES_FATAL_ERROR && sqlstate_of(r) == "0A000",
       "simple LISTEN rejected with 0A000 (got status=%d sqlstate=%s)",
       PQresultStatus(r), sqlstate_of(r).c_str());
    PQclear(r);

    // LISTEN over extended protocol (PQexecParams uses Parse/Bind/Execute) -> same 0A000.
    r = PQexecParams(c.get(), "LISTEN chan2", 0, NULL, NULL, NULL, NULL, 0);
    ok(PQresultStatus(r) == PGRES_FATAL_ERROR && sqlstate_of(r) == "0A000",
       "extended LISTEN rejected with 0A000 (got sqlstate=%s)", sqlstate_of(r).c_str());
    PQclear(r);

    // NOTIFY as a plain query completes cleanly and the connection stays usable.
    PGConnPtr c2 = mk();
    PGresult* rn = PQexec(c2.get(), "NOTIFY chan1, 'hello'");
    bool notify_ok = (PQresultStatus(rn) == PGRES_COMMAND_OK);
    PQclear(rn);
    PGresult* rq = PQexec(c2.get(), "SELECT 1");
    bool still_usable = (PQresultStatus(rq) == PGRES_TUPLES_OK);
    PQclear(rq);
    ok(notify_ok && still_usable, "NOTIFY completes cleanly and connection remains usable");

    return exit_status();
}

  "pgsql-listen_notify_contract-t" : [ "legacy-g4","mysql-auto_increment_delay_multiplex=0-g4","mysql-multiplexing=false-g4","mysql-query_digests=0-g4","mysql-query_digests_keep_comment=1-g4" ],

PROXYSQL31=1 make build_tap_test_debug
WORKSPACE=$(pwd) INFRA_ID=dev-$USER TAP_GROUP=legacy-g4 \
  TEST_PY_TAP_INCL="pgsql-listen_notify_contract-t" test/infra/control/run-tests-isolated.bash

git add test/tap/tests/pgsql-listen_notify_contract-t.cpp test/tap/groups/groups.json
git commit -m "test(pgsql): LISTEN 0A000 rejection + NOTIFY-as-query contract (libpq path)"