From 33340e15ae36ab2d6d216a327283709df9485e92 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 14 Mar 2026 04:50:29 +0000 Subject: [PATCH 1/2] Replace pgwire-lite (libpq-sys) with pure-Rust PG wire client; use reqwest rustls-tls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pgwire-lite v0.1.0 wraps the native libpq C library via libpq-sys, which requires PostgreSQL client headers/libs at build time on every platform: - macOS: 'libpq-fe.h' not found - Linux cross (aarch64): missing libpq / OpenSSL pkg-config - Windows: linker error LNK1181 cannot open 'libpq.lib' Fix 1: Replace pgwire-lite with src/utils/pgwire.rs — a pure-Rust implementation of the PostgreSQL v3 simple-query wire protocol using only std::net::TcpStream. Zero native dependencies. Matches the exact API surface used (PgwireLite::new, query, Value, Notice). Fix 2: Switch reqwest from default native-tls (openssl-sys) to rustls-tls, eliminating the OpenSSL requirement for cross-compiled Linux targets. https://claude.ai/code/session_01GzGtjMcwBXyVW3uKW4F2Ai --- Cargo.lock | 384 +++++++++------------------------------- Cargo.toml | 3 +- src/commands/base.rs | 2 +- src/core/utils.rs | 2 +- src/utils/connection.rs | 4 +- src/utils/mod.rs | 1 + src/utils/pgwire.rs | 322 +++++++++++++++++++++++++++++++++ src/utils/query.rs | 2 +- 8 files changed, 411 insertions(+), 309 deletions(-) create mode 100644 src/utils/pgwire.rs diff --git a/Cargo.lock b/Cargo.lock index 1ff773a..495f3ae 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -135,28 +135,6 @@ version = "1.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "89e25b6adfb930f02d1981565a6e5d9c547ac15a96606256d3b59040e5cd4ca3" -[[package]] -name = "bindgen" -version = "0.64.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4243e6031260db77ede97ad86c27e501d646a27ab57b59a574f725d98ab1fb4" -dependencies = [ - "bitflags 1.3.2", - "cexpr", - "clang-sys", - "lazy_static", - "lazycell", - "log", - "peeking_take_while", - "proc-macro2", - "quote", - "regex", - "rustc-hash", - "shlex", - "syn 1.0.109", - "which", -] - [[package]] name = "bitflags" version = "1.3.2" @@ -237,15 +215,6 @@ dependencies = [ "shlex", ] -[[package]] -name = "cexpr" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766" -dependencies = [ - "nom", -] - [[package]] name = "cfg-if" version = "1.0.0" @@ -298,17 +267,6 @@ dependencies = [ "inout", ] -[[package]] -name = "clang-sys" -version = "1.8.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0b023947811758c97c59bf9d1c188fd619ad4718dcaa767947df1cadb14f39f4" -dependencies = [ - "glob", - "libc", - "libloading", -] - [[package]] name = "clap" version = "4.5.36" @@ -340,7 +298,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn 2.0.100", + "syn", ] [[package]] @@ -519,7 +477,7 @@ checksum = "97369cbbc041bc366949bc74d34658d6cda5621039731c6310521892a3a20ae0" dependencies = [ "proc-macro2", "quote", - "syn 2.0.100", + "syn", ] [[package]] @@ -528,12 +486,6 @@ version = "0.15.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" -[[package]] -name = "either" -version = "1.15.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" - [[package]] name = "encode_unicode" version = "1.0.0" @@ -627,21 +579,6 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" -[[package]] -name = "foreign-types" -version = "0.3.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f6f339eb8adc052cd2ca78910fda869aefa38d22d5cb648e6485e4d3fc06f3b1" -dependencies = [ - "foreign-types-shared", -] - -[[package]] -name = "foreign-types-shared" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "00b0228411908ca8685dba7fc2cdd70ec9990a6e753e89b6ac91a84c40fbaf4b" - [[package]] name = "form_urlencoded" version = "1.2.1" @@ -738,12 +675,6 @@ version = "0.31.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "07e28edb80900c19c28f1072f2e8aeca7fa06b23cd4169cefe1af5aa3260783f" -[[package]] -name = "glob" -version = "0.3.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a8d1add55171497b4705a648c6b583acafb01d58050a51727785f0b2c8e0a2b2" - [[package]] name = "globset" version = "0.4.16" @@ -814,15 +745,6 @@ dependencies = [ "digest", ] -[[package]] -name = "home" -version = "0.5.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "589533453244b0995c858700322199b2becb13b627df2851f64a2775d024abcf" -dependencies = [ - "windows-sys 0.59.0", -] - [[package]] name = "http" version = "0.2.12" @@ -897,16 +819,17 @@ dependencies = [ ] [[package]] -name = "hyper-tls" -version = "0.5.0" +name = "hyper-rustls" +version = "0.24.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d6183ddfa99b85da61a140bea0efc93fdf56ceaa041b37d553518030827f9905" +checksum = "ec3efd23720e2049821a693cbc7e65ea87c72f1c58ff2f9522ff332b1491e590" dependencies = [ - "bytes", + "futures-util", + "http", "hyper", - "native-tls", + "rustls", "tokio", - "tokio-native-tls", + "tokio-rustls", ] [[package]] @@ -1048,7 +971,7 @@ checksum = "1ec89e9337638ecdc08744df490b221a7399bf8d164eb52a665454e60e075ad6" dependencies = [ "proc-macro2", "quote", - "syn 2.0.100", + "syn", ] [[package]] @@ -1175,58 +1098,18 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" -[[package]] -name = "lazycell" -version = "1.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55" - [[package]] name = "libc" version = "0.2.172" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d750af042f7ef4f724306de029d18836c26c1765a54a6a3f094cbd23a7267ffa" -[[package]] -name = "libloading" -version = "0.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fc2f4eb4bc735547cfed7c0a4922cbd04a4655978c09b54f1f7b228750664c34" -dependencies = [ - "cfg-if", - "windows-targets 0.52.6", -] - [[package]] name = "libm" version = "0.2.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8355be11b20d696c8f18f6cc018c4e372165b1fa8126cef092399c9951984ffa" -[[package]] -name = "libpq" -version = "4.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "57eb9f8893722a29eab34ec11b42a0455abf265162871cf5d6fa4f04842b8fc5" -dependencies = [ - "bitflags 2.9.0", - "libc", - "libpq-sys", - "log", - "thiserror 1.0.69", -] - -[[package]] -name = "libpq-sys" -version = "0.8.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2ef060ac05c207c85da15f4eb629100c8782e0db4c06a3c91c86be9c18ae8a23" -dependencies = [ - "bindgen", - "pkg-config", - "vcpkg", -] - [[package]] name = "libredox" version = "0.1.3" @@ -1273,12 +1156,6 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" -[[package]] -name = "minimal-lexical" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" - [[package]] name = "miniz_oxide" version = "0.8.8" @@ -1299,23 +1176,6 @@ dependencies = [ "windows-sys 0.52.0", ] -[[package]] -name = "native-tls" -version = "0.2.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "87de3442987e9dbec73158d5c715e7ad9072fda936bb03d19d7fa10e00520f0e" -dependencies = [ - "libc", - "log", - "openssl", - "openssl-probe", - "openssl-sys", - "schannel", - "security-framework", - "security-framework-sys", - "tempfile", -] - [[package]] name = "nibble_vec" version = "0.1.0" @@ -1337,16 +1197,6 @@ dependencies = [ "libc", ] -[[package]] -name = "nom" -version = "7.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" -dependencies = [ - "memchr", - "minimal-lexical", -] - [[package]] name = "num-conv" version = "0.1.0" @@ -1383,50 +1233,6 @@ version = "1.21.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d" -[[package]] -name = "openssl" -version = "0.10.72" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fedfea7d58a1f73118430a55da6a286e7b044961736ce96a16a17068ea25e5da" -dependencies = [ - "bitflags 2.9.0", - "cfg-if", - "foreign-types", - "libc", - "once_cell", - "openssl-macros", - "openssl-sys", -] - -[[package]] -name = "openssl-macros" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.100", -] - -[[package]] -name = "openssl-probe" -version = "0.1.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e" - -[[package]] -name = "openssl-sys" -version = "0.9.107" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8288979acd84749c744a9014b4382d42b8f7b2592847b5afb2ed29e5d16ede07" -dependencies = [ - "cc", - "libc", - "pkg-config", - "vcpkg", -] - [[package]] name = "parse-zoneinfo" version = "0.3.1" @@ -1459,12 +1265,6 @@ dependencies = [ "sha2", ] -[[package]] -name = "peeking_take_while" -version = "0.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "19b17cddbe7ec3f8bc800887bab5e717348c95ea2ca0b1bf0837fb964dc67099" - [[package]] name = "percent-encoding" version = "2.3.1" @@ -1502,7 +1302,7 @@ dependencies = [ "pest_meta", "proc-macro2", "quote", - "syn 2.0.100", + "syn", ] [[package]] @@ -1516,17 +1316,6 @@ dependencies = [ "sha2", ] -[[package]] -name = "pgwire-lite" -version = "0.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85b08a19f39360a988ed911d66fd586f5c03f14252618b62941cc9af061456c0" -dependencies = [ - "libpq", - "libpq-sys", - "log", -] - [[package]] name = "phf" version = "0.11.3" @@ -1723,15 +1512,15 @@ dependencies = [ "http", "http-body", "hyper", - "hyper-tls", + "hyper-rustls", "ipnet", "js-sys", "log", "mime", - "native-tls", "once_cell", "percent-encoding", "pin-project-lite", + "rustls", "rustls-pemfile", "serde", "serde_json", @@ -1739,26 +1528,35 @@ dependencies = [ "sync_wrapper", "system-configuration", "tokio", - "tokio-native-tls", + "tokio-rustls", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", "web-sys", + "webpki-roots", "winreg", ] [[package]] -name = "rustc-demangle" -version = "0.1.24" +name = "ring" +version = "0.17.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "719b953e2095829ee67db738b3bfa9fa368c94900df327b3f07fe6e794d2fe1f" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.15", + "libc", + "untrusted", + "windows-sys 0.52.0", +] [[package]] -name = "rustc-hash" -version = "1.1.0" +name = "rustc-demangle" +version = "0.1.24" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2" +checksum = "719b953e2095829ee67db738b3bfa9fa368c94900df327b3f07fe6e794d2fe1f" [[package]] name = "rustix" @@ -1786,6 +1584,18 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "rustls" +version = "0.21.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f56a14d1f48b391359b22f731fd4bd7e43c97f3c50eee276f3aa09c94784d3e" +dependencies = [ + "log", + "ring", + "rustls-webpki", + "sct", +] + [[package]] name = "rustls-pemfile" version = "1.0.4" @@ -1795,6 +1605,16 @@ dependencies = [ "base64", ] +[[package]] +name = "rustls-webpki" +version = "0.101.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b6275d1ee7a1cd780b64aca7726599a1dbc893b1e64144529e55c3c2f745765" +dependencies = [ + "ring", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.20" @@ -1839,15 +1659,6 @@ dependencies = [ "winapi-util", ] -[[package]] -name = "schannel" -version = "0.1.27" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1f29ebaa345f945cec9fbbc532eb307f0fdad8161f281b6369539c8d84876b3d" -dependencies = [ - "windows-sys 0.59.0", -] - [[package]] name = "scopeguard" version = "1.2.0" @@ -1855,26 +1666,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" [[package]] -name = "security-framework" -version = "2.11.1" +name = "sct" +version = "0.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "897b2245f0b511c87893af39b033e5ca9cce68824c4d7e7630b5a1d339658d02" +checksum = "da046153aa2352493d6cb7da4b6e5c0c057d8a1d0a9aa8560baffdd945acd414" dependencies = [ - "bitflags 2.9.0", - "core-foundation", - "core-foundation-sys", - "libc", - "security-framework-sys", -] - -[[package]] -name = "security-framework-sys" -version = "2.14.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49db231d56a190491cb4aeda9527f1ad45345af50b0851622a7adb8c03b01c32" -dependencies = [ - "core-foundation-sys", - "libc", + "ring", + "untrusted", ] [[package]] @@ -1894,7 +1692,7 @@ checksum = "5b0276cf7f2c73365f7157c8123c21cd9a50fbbd844757af28ca1f5925fc2a00" dependencies = [ "proc-macro2", "quote", - "syn 2.0.100", + "syn", ] [[package]] @@ -2011,7 +1809,7 @@ checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3" [[package]] name = "stackql-deploy" -version = "0.1.0" +version = "2.0.0" dependencies = [ "base64", "chrono", @@ -2022,7 +1820,6 @@ dependencies = [ "indicatif", "log", "once_cell", - "pgwire-lite", "regex", "reqwest", "rustyline", @@ -2055,17 +1852,6 @@ version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" -[[package]] -name = "syn" -version = "1.0.109" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "72b64191b275b66ffe2469e8af2c1cfe3bafa67b529ead792a6d0160888b4237" -dependencies = [ - "proc-macro2", - "quote", - "unicode-ident", -] - [[package]] name = "syn" version = "2.0.100" @@ -2091,7 +1877,7 @@ checksum = "c8af7666ab7b6390ab78131fb5b0fce11d6b7a6951602017c35fa82800708971" dependencies = [ "proc-macro2", "quote", - "syn 2.0.100", + "syn", ] [[package]] @@ -2185,7 +1971,7 @@ checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" dependencies = [ "proc-macro2", "quote", - "syn 2.0.100", + "syn", ] [[package]] @@ -2196,7 +1982,7 @@ checksum = "7f7cf42b4507d8ea322120659672cf1b9dbb93f8f2d4ecfd6e51350ff5b17a1d" dependencies = [ "proc-macro2", "quote", - "syn 2.0.100", + "syn", ] [[package]] @@ -2244,12 +2030,12 @@ dependencies = [ ] [[package]] -name = "tokio-native-tls" -version = "0.3.1" +name = "tokio-rustls" +version = "0.24.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bbae76ab933c85776efabc971569dd6119c580d8f5d448769dec1764bf796ef2" +checksum = "c28327cf380ac148141087fbfb9de9d7bd4e84ab5d2c28fbc911d753de8a7081" dependencies = [ - "native-tls", + "rustls", "tokio", ] @@ -2389,6 +2175,12 @@ version = "0.2.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861" +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + [[package]] name = "url" version = "2.5.4" @@ -2429,12 +2221,6 @@ dependencies = [ "wasm-bindgen", ] -[[package]] -name = "vcpkg" -version = "0.2.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" - [[package]] name = "version_check" version = "0.9.5" @@ -2497,7 +2283,7 @@ dependencies = [ "log", "proc-macro2", "quote", - "syn 2.0.100", + "syn", "wasm-bindgen-shared", ] @@ -2532,7 +2318,7 @@ checksum = "8ae87ea40c9f689fc23f209965b6fb8a99ad69aeeb0231408be24920604395de" dependencies = [ "proc-macro2", "quote", - "syn 2.0.100", + "syn", "wasm-bindgen-backend", "wasm-bindgen-shared", ] @@ -2567,16 +2353,10 @@ dependencies = [ ] [[package]] -name = "which" -version = "4.4.2" +name = "webpki-roots" +version = "0.25.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "87ba24419a2078cd2b0f2ede2691b6c66d8e47836da3b6db8265ebad47afbfc7" -dependencies = [ - "either", - "home", - "once_cell", - "rustix 0.38.44", -] +checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1" [[package]] name = "winapi" @@ -2630,7 +2410,7 @@ checksum = "a47fddd13af08290e67f4acabf4b459f647552718f683a7b415d290ac744a836" dependencies = [ "proc-macro2", "quote", - "syn 2.0.100", + "syn", ] [[package]] @@ -2641,7 +2421,7 @@ checksum = "bd9211b69f8dcdfa817bfd14bf1c97c9188afa36f4750130fcdf3f400eca9fa8" dependencies = [ "proc-macro2", "quote", - "syn 2.0.100", + "syn", ] [[package]] @@ -2867,7 +2647,7 @@ checksum = "2380878cad4ac9aac1e2435f3eb4020e8374b5f13c296cb75b4620ff8e229154" dependencies = [ "proc-macro2", "quote", - "syn 2.0.100", + "syn", "synstructure", ] @@ -2888,7 +2668,7 @@ checksum = "a996a8f63c5c4448cd959ac1bab0aaa3306ccfd060472f85943ee0750f0169be" dependencies = [ "proc-macro2", "quote", - "syn 2.0.100", + "syn", ] [[package]] @@ -2908,7 +2688,7 @@ checksum = "d71e5d6e06ab090c67b5e44993ec16b72dcbaabc526db883a360057678b48502" dependencies = [ "proc-macro2", "quote", - "syn 2.0.100", + "syn", "synstructure", ] @@ -2931,7 +2711,7 @@ checksum = "6eafa6dfb17584ea3e2bd6e76e0cc15ad7af12b09abdd1ca55961bed9b1063c6" dependencies = [ "proc-macro2", "quote", - "syn 2.0.100", + "syn", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 48e0cea..dce2d51 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,9 +24,8 @@ rustyline = "10.0" tera = "1.19.0" log = "0.4" env_logger = "0.10" -pgwire-lite = "0.1.0" zip = "0.6" -reqwest = { version = "0.11", features = ["blocking", "json"] } +reqwest = { version = "0.11", default-features = false, features = ["blocking", "json", "rustls-tls"] } indicatif = "0.17" unicode-width = "0.1.10" once_cell = "1.17.0" diff --git a/src/commands/base.rs b/src/commands/base.rs index 6690d72..e4da698 100644 --- a/src/commands/base.rs +++ b/src/commands/base.rs @@ -11,7 +11,7 @@ use std::path::Path; use std::process; use log::{debug, error, info}; -use pgwire_lite::PgwireLite; +use crate::utils::pgwire::PgwireLite; use crate::core::config::{get_full_context, render_globals, render_string_value}; use crate::core::env::load_env_vars; diff --git a/src/core/utils.rs b/src/core/utils.rs index 0e2d6f0..e10903b 100644 --- a/src/core/utils.rs +++ b/src/core/utils.rs @@ -12,7 +12,7 @@ use std::thread; use std::time::{Duration, Instant}; use log::{debug, error, info, warn}; -use pgwire_lite::PgwireLite; +use crate::utils::pgwire::PgwireLite; use crate::utils::query::{execute_query, QueryResult}; diff --git a/src/utils/connection.rs b/src/utils/connection.rs index d3ddeee..7b8d18e 100644 --- a/src/utils/connection.rs +++ b/src/utils/connection.rs @@ -21,9 +21,9 @@ use std::process; use colored::*; -use pgwire_lite::PgwireLite; use crate::globals::{server_host, server_port}; +use crate::utils::pgwire::PgwireLite; /// Creates a new PgwireLite client connection pub fn create_client() -> PgwireLite { @@ -38,7 +38,7 @@ pub fn create_client() -> PgwireLite { }); println!("Connected to stackql server at {}:{}", host, port); - println!("Using libpq version: {}", client.libpq_version()); + println!("Using pgwire client: {}", client.libpq_version()); client } diff --git a/src/utils/mod.rs b/src/utils/mod.rs index 7af634d..7622b7b 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -3,6 +3,7 @@ pub mod connection; pub mod display; pub mod download; pub mod logging; +pub mod pgwire; pub mod platform; pub mod query; pub mod server; diff --git a/src/utils/pgwire.rs b/src/utils/pgwire.rs new file mode 100644 index 0000000..17c1b86 --- /dev/null +++ b/src/utils/pgwire.rs @@ -0,0 +1,322 @@ +// utils/pgwire.rs + +//! Pure-Rust PostgreSQL simple-query wire protocol client. +//! +//! Implements only what stackql-deploy needs: unencrypted TCP connections +//! to a local StackQL server using the PostgreSQL simple query protocol (v3). +//! No native dependencies (replaces pgwire-lite → libpq-sys). + +use std::collections::HashMap; +use std::io::{Read, Write}; +use std::net::TcpStream; + +/// A single column value returned from a query. +pub enum Value { + String(String), + Null, + Bool(bool), + Integer(i64), + Float(f64), + Bytes(Vec), +} + +/// A server notice (NOTICE, WARNING, etc.). +pub struct Notice { + pub fields: HashMap, +} + +/// The result of a [`PgwireLite::query`] call. +pub struct PgQueryResult { + pub column_names: Vec, + pub rows: Vec>, + pub notices: Vec, + /// Row count reported by CommandComplete (INSERT/UPDATE/DELETE n). + pub row_count: usize, +} + +/// Minimal PostgreSQL wire-protocol client. +pub struct PgwireLite { + stream: TcpStream, +} + +impl PgwireLite { + /// Connect to a PostgreSQL-protocol server (e.g. StackQL) at `host:port`. + /// + /// `_ssl` and `_verbosity` are accepted for API compatibility but ignored; + /// the connection is always unencrypted (StackQL default). + pub fn new(host: &str, port: u16, _ssl: bool, _verbosity: &str) -> Result { + let addr = format!("{}:{}", host, port); + let stream = TcpStream::connect(&addr) + .map_err(|e| format!("Connection to {} failed: {}", addr, e))?; + + let mut client = PgwireLite { stream }; + client.startup()?; + Ok(client) + } + + /// Returns a version string (no libpq; just identifies the client). + pub fn libpq_version(&self) -> String { + "pure-rust-pgwire-client".to_string() + } + + // ------------------------------------------------------------------ + // Startup handshake + // ------------------------------------------------------------------ + + fn startup(&mut self) -> Result<(), String> { + // Protocol version 3.0 = 0x00_03_00_00 + const PROTOCOL_V3: i32 = 196608; + + // Startup message: user=stackql, database=stackql, then double-null + let params = b"user\0stackql\0database\0stackql\0\0"; + let total_len = 4 + 4 + params.len(); // length field + protocol + params + + let mut msg = Vec::with_capacity(total_len); + msg.extend_from_slice(&(total_len as i32).to_be_bytes()); + msg.extend_from_slice(&PROTOCOL_V3.to_be_bytes()); + msg.extend_from_slice(params); + + self.stream + .write_all(&msg) + .map_err(|e| format!("Startup write error: {}", e))?; + + // Process auth / parameter-status messages until ReadyForQuery + loop { + let msg_type = self.read_byte()?; + let payload_len = self.read_i32()? as usize; + // payload_len includes the 4 bytes of the length field itself + let data = self.read_bytes(payload_len.saturating_sub(4))?; + + match msg_type { + b'R' => { + // AuthenticationRequest + let auth_type = + i32::from_be_bytes(data[..4].try_into().map_err(|_| "Bad auth")?); + if auth_type != 0 { + return Err(format!( + "Unsupported authentication type {} from server", + auth_type + )); + } + // AuthenticationOk — nothing to do + } + b'K' => {} // BackendKeyData — ignore + b'S' => {} // ParameterStatus — ignore + b'Z' => break, // ReadyForQuery + b'E' => return Err(parse_error_fields(&data)), + b'N' => {} // NoticeResponse during startup — ignore + _ => {} // Unknown message type — skip + } + } + + Ok(()) + } + + // ------------------------------------------------------------------ + // Query + // ------------------------------------------------------------------ + + /// Execute a simple (non-prepared) SQL query and return structured results. + pub fn query(&mut self, sql: &str) -> Result { + // Send Query message: 'Q' | int32(len) | sql\0 + let sql_bytes = sql.as_bytes(); + let payload_len = 4 + sql_bytes.len() + 1; // length field + sql + null + + let mut msg = Vec::with_capacity(1 + payload_len); + msg.push(b'Q'); + msg.extend_from_slice(&(payload_len as i32).to_be_bytes()); + msg.extend_from_slice(sql_bytes); + msg.push(0u8); + + self.stream + .write_all(&msg) + .map_err(|e| format!("Query write error: {}", e))?; + + // Collect response messages + let mut column_names: Vec = Vec::new(); + let mut rows: Vec> = Vec::new(); + let mut notices: Vec = Vec::new(); + let mut row_count: usize = 0; + + loop { + let msg_type = self.read_byte()?; + let payload_len = self.read_i32()? as usize; + let data = self.read_bytes(payload_len.saturating_sub(4))?; + + match msg_type { + b'T' => { + // RowDescription + column_names = parse_row_description(&data); + } + b'D' => { + // DataRow + let row = parse_data_row(&data, &column_names); + rows.push(row); + } + b'C' => { + // CommandComplete — tag like "SELECT 5", "INSERT 0 1", "UPDATE 3" + let tag = std::str::from_utf8(data.strip_suffix(b"\0").unwrap_or(&data)) + .unwrap_or("") + .to_string(); + if let Some(n) = tag.split_whitespace().last().and_then(|s| s.parse().ok()) { + row_count = n; + } + } + b'N' => { + notices.push(parse_notice_fields(&data)); + } + b'E' => { + return Err(parse_error_fields(&data)); + } + b'I' => {} // EmptyQueryResponse + b'Z' => break, // ReadyForQuery — done + _ => {} + } + } + + Ok(PgQueryResult { + column_names, + rows, + notices, + row_count, + }) + } + + // ------------------------------------------------------------------ + // Low-level I/O helpers + // ------------------------------------------------------------------ + + fn read_byte(&mut self) -> Result { + let mut buf = [0u8; 1]; + self.stream + .read_exact(&mut buf) + .map_err(|e| format!("Read error: {}", e))?; + Ok(buf[0]) + } + + fn read_i32(&mut self) -> Result { + let mut buf = [0u8; 4]; + self.stream + .read_exact(&mut buf) + .map_err(|e| format!("Read error: {}", e))?; + Ok(i32::from_be_bytes(buf)) + } + + fn read_bytes(&mut self, n: usize) -> Result, String> { + let mut buf = vec![0u8; n]; + self.stream + .read_exact(&mut buf) + .map_err(|e| format!("Read error: {}", e))?; + Ok(buf) + } +} + +// ------------------------------------------------------------------ +// Message parsers (free functions for readability) +// ------------------------------------------------------------------ + +fn parse_row_description(data: &[u8]) -> Vec { + let mut names = Vec::new(); + if data.len() < 2 { + return names; + } + let num_fields = u16::from_be_bytes([data[0], data[1]]) as usize; + let mut pos = 2; + + for _ in 0..num_fields { + // Null-terminated field name + let Some(null_off) = data[pos..].iter().position(|&b| b == 0) else { + break; + }; + let name = String::from_utf8_lossy(&data[pos..pos + null_off]).into_owned(); + names.push(name); + // Skip: name + null(1) + tableOID(4) + attrNum(2) + typeOID(4) + typeSize(2) + // + typeMod(4) + formatCode(2) = 19 bytes after the null + pos += null_off + 1 + 18; + } + names +} + +fn parse_data_row(data: &[u8], columns: &[String]) -> HashMap { + let mut row = HashMap::new(); + if data.len() < 2 { + return row; + } + let num_cols = u16::from_be_bytes([data[0], data[1]]) as usize; + let mut pos = 2; + + for i in 0..num_cols.min(columns.len()) { + if pos + 4 > data.len() { + break; + } + let col_len = i32::from_be_bytes([data[pos], data[pos + 1], data[pos + 2], data[pos + 3]]); + pos += 4; + + let value = if col_len < 0 { + Value::Null + } else { + let len = col_len as usize; + if pos + len > data.len() { + break; + } + let s = String::from_utf8_lossy(&data[pos..pos + len]).into_owned(); + pos += len; + Value::String(s) + }; + + row.insert(columns[i].clone(), value); + } + row +} + +fn parse_notice_fields(data: &[u8]) -> Notice { + let mut fields = HashMap::new(); + let mut pos = 0; + + while pos < data.len() { + let field_code = data[pos]; + pos += 1; + if field_code == 0 { + break; + } + let Some(null_off) = data[pos..].iter().position(|&b| b == 0) else { + break; + }; + let value = String::from_utf8_lossy(&data[pos..pos + null_off]).into_owned(); + pos += null_off + 1; + + let key = match field_code { + b'S' => "severity", + b'M' => "message", + b'D' => "detail", + b'H' => "hint", + b'C' => "code", + b'P' => "position", + b'W' => "where", + _ => continue, + }; + fields.insert(key.to_string(), value); + } + + Notice { fields } +} + +fn parse_error_fields(data: &[u8]) -> String { + let mut pos = 0; + while pos < data.len() { + let field_code = data[pos]; + pos += 1; + if field_code == 0 { + break; + } + let Some(null_off) = data[pos..].iter().position(|&b| b == 0) else { + break; + }; + let value = String::from_utf8_lossy(&data[pos..pos + null_off]).into_owned(); + pos += null_off + 1; + if field_code == b'M' { + return value; + } + } + "Unknown server error".to_string() +} diff --git a/src/utils/query.rs b/src/utils/query.rs index 5159d53..ccc6766 100644 --- a/src/utils/query.rs +++ b/src/utils/query.rs @@ -26,7 +26,7 @@ //! } //! ``` -use pgwire_lite::{PgwireLite, Value}; +use crate::utils::pgwire::{PgwireLite, Value}; /// Represents a column in a query result. pub struct QueryResultColumn { From 8f29b6030e4e97a689ea698b7073eb4d2f8715c3 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 14 Mar 2026 04:54:48 +0000 Subject: [PATCH 2/2] Fix cargo fmt and clippy warnings in pgwire client and import ordering - Reorder crate:: imports in base.rs and core/utils.rs to satisfy rustfmt (external crates before local crate imports, within their own groups) - Fix trailing comment alignment in pgwire.rs match arms per rustfmt rules - Replace needless range loop with iterator in parse_data_row (clippy) https://claude.ai/code/session_01GzGtjMcwBXyVW3uKW4F2Ai --- src/commands/base.rs | 2 +- src/core/utils.rs | 2 +- src/utils/pgwire.rs | 644 +++++++++++++++++++++---------------------- 3 files changed, 324 insertions(+), 324 deletions(-) diff --git a/src/commands/base.rs b/src/commands/base.rs index e4da698..9c14573 100644 --- a/src/commands/base.rs +++ b/src/commands/base.rs @@ -11,7 +11,6 @@ use std::path::Path; use std::process; use log::{debug, error, info}; -use crate::utils::pgwire::PgwireLite; use crate::core::config::{get_full_context, render_globals, render_string_value}; use crate::core::env::load_env_vars; @@ -25,6 +24,7 @@ use crate::core::utils::{ use crate::resource::manifest::{Manifest, Resource}; use crate::resource::validation::validate_manifest; use crate::template::engine::TemplateEngine; +use crate::utils::pgwire::PgwireLite; /// Core state for all command operations, equivalent to Python's StackQLBase. pub struct CommandRunner { diff --git a/src/core/utils.rs b/src/core/utils.rs index e10903b..80bd5aa 100644 --- a/src/core/utils.rs +++ b/src/core/utils.rs @@ -12,8 +12,8 @@ use std::thread; use std::time::{Duration, Instant}; use log::{debug, error, info, warn}; -use crate::utils::pgwire::PgwireLite; +use crate::utils::pgwire::PgwireLite; use crate::utils::query::{execute_query, QueryResult}; /// Exit with error message. Matches Python's `catch_error_and_exit`. diff --git a/src/utils/pgwire.rs b/src/utils/pgwire.rs index 17c1b86..b9453c8 100644 --- a/src/utils/pgwire.rs +++ b/src/utils/pgwire.rs @@ -1,322 +1,322 @@ -// utils/pgwire.rs - -//! Pure-Rust PostgreSQL simple-query wire protocol client. -//! -//! Implements only what stackql-deploy needs: unencrypted TCP connections -//! to a local StackQL server using the PostgreSQL simple query protocol (v3). -//! No native dependencies (replaces pgwire-lite → libpq-sys). - -use std::collections::HashMap; -use std::io::{Read, Write}; -use std::net::TcpStream; - -/// A single column value returned from a query. -pub enum Value { - String(String), - Null, - Bool(bool), - Integer(i64), - Float(f64), - Bytes(Vec), -} - -/// A server notice (NOTICE, WARNING, etc.). -pub struct Notice { - pub fields: HashMap, -} - -/// The result of a [`PgwireLite::query`] call. -pub struct PgQueryResult { - pub column_names: Vec, - pub rows: Vec>, - pub notices: Vec, - /// Row count reported by CommandComplete (INSERT/UPDATE/DELETE n). - pub row_count: usize, -} - -/// Minimal PostgreSQL wire-protocol client. -pub struct PgwireLite { - stream: TcpStream, -} - -impl PgwireLite { - /// Connect to a PostgreSQL-protocol server (e.g. StackQL) at `host:port`. - /// - /// `_ssl` and `_verbosity` are accepted for API compatibility but ignored; - /// the connection is always unencrypted (StackQL default). - pub fn new(host: &str, port: u16, _ssl: bool, _verbosity: &str) -> Result { - let addr = format!("{}:{}", host, port); - let stream = TcpStream::connect(&addr) - .map_err(|e| format!("Connection to {} failed: {}", addr, e))?; - - let mut client = PgwireLite { stream }; - client.startup()?; - Ok(client) - } - - /// Returns a version string (no libpq; just identifies the client). - pub fn libpq_version(&self) -> String { - "pure-rust-pgwire-client".to_string() - } - - // ------------------------------------------------------------------ - // Startup handshake - // ------------------------------------------------------------------ - - fn startup(&mut self) -> Result<(), String> { - // Protocol version 3.0 = 0x00_03_00_00 - const PROTOCOL_V3: i32 = 196608; - - // Startup message: user=stackql, database=stackql, then double-null - let params = b"user\0stackql\0database\0stackql\0\0"; - let total_len = 4 + 4 + params.len(); // length field + protocol + params - - let mut msg = Vec::with_capacity(total_len); - msg.extend_from_slice(&(total_len as i32).to_be_bytes()); - msg.extend_from_slice(&PROTOCOL_V3.to_be_bytes()); - msg.extend_from_slice(params); - - self.stream - .write_all(&msg) - .map_err(|e| format!("Startup write error: {}", e))?; - - // Process auth / parameter-status messages until ReadyForQuery - loop { - let msg_type = self.read_byte()?; - let payload_len = self.read_i32()? as usize; - // payload_len includes the 4 bytes of the length field itself - let data = self.read_bytes(payload_len.saturating_sub(4))?; - - match msg_type { - b'R' => { - // AuthenticationRequest - let auth_type = - i32::from_be_bytes(data[..4].try_into().map_err(|_| "Bad auth")?); - if auth_type != 0 { - return Err(format!( - "Unsupported authentication type {} from server", - auth_type - )); - } - // AuthenticationOk — nothing to do - } - b'K' => {} // BackendKeyData — ignore - b'S' => {} // ParameterStatus — ignore - b'Z' => break, // ReadyForQuery - b'E' => return Err(parse_error_fields(&data)), - b'N' => {} // NoticeResponse during startup — ignore - _ => {} // Unknown message type — skip - } - } - - Ok(()) - } - - // ------------------------------------------------------------------ - // Query - // ------------------------------------------------------------------ - - /// Execute a simple (non-prepared) SQL query and return structured results. - pub fn query(&mut self, sql: &str) -> Result { - // Send Query message: 'Q' | int32(len) | sql\0 - let sql_bytes = sql.as_bytes(); - let payload_len = 4 + sql_bytes.len() + 1; // length field + sql + null - - let mut msg = Vec::with_capacity(1 + payload_len); - msg.push(b'Q'); - msg.extend_from_slice(&(payload_len as i32).to_be_bytes()); - msg.extend_from_slice(sql_bytes); - msg.push(0u8); - - self.stream - .write_all(&msg) - .map_err(|e| format!("Query write error: {}", e))?; - - // Collect response messages - let mut column_names: Vec = Vec::new(); - let mut rows: Vec> = Vec::new(); - let mut notices: Vec = Vec::new(); - let mut row_count: usize = 0; - - loop { - let msg_type = self.read_byte()?; - let payload_len = self.read_i32()? as usize; - let data = self.read_bytes(payload_len.saturating_sub(4))?; - - match msg_type { - b'T' => { - // RowDescription - column_names = parse_row_description(&data); - } - b'D' => { - // DataRow - let row = parse_data_row(&data, &column_names); - rows.push(row); - } - b'C' => { - // CommandComplete — tag like "SELECT 5", "INSERT 0 1", "UPDATE 3" - let tag = std::str::from_utf8(data.strip_suffix(b"\0").unwrap_or(&data)) - .unwrap_or("") - .to_string(); - if let Some(n) = tag.split_whitespace().last().and_then(|s| s.parse().ok()) { - row_count = n; - } - } - b'N' => { - notices.push(parse_notice_fields(&data)); - } - b'E' => { - return Err(parse_error_fields(&data)); - } - b'I' => {} // EmptyQueryResponse - b'Z' => break, // ReadyForQuery — done - _ => {} - } - } - - Ok(PgQueryResult { - column_names, - rows, - notices, - row_count, - }) - } - - // ------------------------------------------------------------------ - // Low-level I/O helpers - // ------------------------------------------------------------------ - - fn read_byte(&mut self) -> Result { - let mut buf = [0u8; 1]; - self.stream - .read_exact(&mut buf) - .map_err(|e| format!("Read error: {}", e))?; - Ok(buf[0]) - } - - fn read_i32(&mut self) -> Result { - let mut buf = [0u8; 4]; - self.stream - .read_exact(&mut buf) - .map_err(|e| format!("Read error: {}", e))?; - Ok(i32::from_be_bytes(buf)) - } - - fn read_bytes(&mut self, n: usize) -> Result, String> { - let mut buf = vec![0u8; n]; - self.stream - .read_exact(&mut buf) - .map_err(|e| format!("Read error: {}", e))?; - Ok(buf) - } -} - -// ------------------------------------------------------------------ -// Message parsers (free functions for readability) -// ------------------------------------------------------------------ - -fn parse_row_description(data: &[u8]) -> Vec { - let mut names = Vec::new(); - if data.len() < 2 { - return names; - } - let num_fields = u16::from_be_bytes([data[0], data[1]]) as usize; - let mut pos = 2; - - for _ in 0..num_fields { - // Null-terminated field name - let Some(null_off) = data[pos..].iter().position(|&b| b == 0) else { - break; - }; - let name = String::from_utf8_lossy(&data[pos..pos + null_off]).into_owned(); - names.push(name); - // Skip: name + null(1) + tableOID(4) + attrNum(2) + typeOID(4) + typeSize(2) - // + typeMod(4) + formatCode(2) = 19 bytes after the null - pos += null_off + 1 + 18; - } - names -} - -fn parse_data_row(data: &[u8], columns: &[String]) -> HashMap { - let mut row = HashMap::new(); - if data.len() < 2 { - return row; - } - let num_cols = u16::from_be_bytes([data[0], data[1]]) as usize; - let mut pos = 2; - - for i in 0..num_cols.min(columns.len()) { - if pos + 4 > data.len() { - break; - } - let col_len = i32::from_be_bytes([data[pos], data[pos + 1], data[pos + 2], data[pos + 3]]); - pos += 4; - - let value = if col_len < 0 { - Value::Null - } else { - let len = col_len as usize; - if pos + len > data.len() { - break; - } - let s = String::from_utf8_lossy(&data[pos..pos + len]).into_owned(); - pos += len; - Value::String(s) - }; - - row.insert(columns[i].clone(), value); - } - row -} - -fn parse_notice_fields(data: &[u8]) -> Notice { - let mut fields = HashMap::new(); - let mut pos = 0; - - while pos < data.len() { - let field_code = data[pos]; - pos += 1; - if field_code == 0 { - break; - } - let Some(null_off) = data[pos..].iter().position(|&b| b == 0) else { - break; - }; - let value = String::from_utf8_lossy(&data[pos..pos + null_off]).into_owned(); - pos += null_off + 1; - - let key = match field_code { - b'S' => "severity", - b'M' => "message", - b'D' => "detail", - b'H' => "hint", - b'C' => "code", - b'P' => "position", - b'W' => "where", - _ => continue, - }; - fields.insert(key.to_string(), value); - } - - Notice { fields } -} - -fn parse_error_fields(data: &[u8]) -> String { - let mut pos = 0; - while pos < data.len() { - let field_code = data[pos]; - pos += 1; - if field_code == 0 { - break; - } - let Some(null_off) = data[pos..].iter().position(|&b| b == 0) else { - break; - }; - let value = String::from_utf8_lossy(&data[pos..pos + null_off]).into_owned(); - pos += null_off + 1; - if field_code == b'M' { - return value; - } - } - "Unknown server error".to_string() -} +// utils/pgwire.rs + +//! Pure-Rust PostgreSQL simple-query wire protocol client. +//! +//! Implements only what stackql-deploy needs: unencrypted TCP connections +//! to a local StackQL server using the PostgreSQL simple query protocol (v3). +//! No native dependencies (replaces pgwire-lite → libpq-sys). + +use std::collections::HashMap; +use std::io::{Read, Write}; +use std::net::TcpStream; + +/// A single column value returned from a query. +pub enum Value { + String(String), + Null, + Bool(bool), + Integer(i64), + Float(f64), + Bytes(Vec), +} + +/// A server notice (NOTICE, WARNING, etc.). +pub struct Notice { + pub fields: HashMap, +} + +/// The result of a [`PgwireLite::query`] call. +pub struct PgQueryResult { + pub column_names: Vec, + pub rows: Vec>, + pub notices: Vec, + /// Row count reported by CommandComplete (INSERT/UPDATE/DELETE n). + pub row_count: usize, +} + +/// Minimal PostgreSQL wire-protocol client. +pub struct PgwireLite { + stream: TcpStream, +} + +impl PgwireLite { + /// Connect to a PostgreSQL-protocol server (e.g. StackQL) at `host:port`. + /// + /// `_ssl` and `_verbosity` are accepted for API compatibility but ignored; + /// the connection is always unencrypted (StackQL default). + pub fn new(host: &str, port: u16, _ssl: bool, _verbosity: &str) -> Result { + let addr = format!("{}:{}", host, port); + let stream = TcpStream::connect(&addr) + .map_err(|e| format!("Connection to {} failed: {}", addr, e))?; + + let mut client = PgwireLite { stream }; + client.startup()?; + Ok(client) + } + + /// Returns a version string (no libpq; just identifies the client). + pub fn libpq_version(&self) -> String { + "pure-rust-pgwire-client".to_string() + } + + // ------------------------------------------------------------------ + // Startup handshake + // ------------------------------------------------------------------ + + fn startup(&mut self) -> Result<(), String> { + // Protocol version 3.0 = 0x00_03_00_00 + const PROTOCOL_V3: i32 = 196608; + + // Startup message: user=stackql, database=stackql, then double-null + let params = b"user\0stackql\0database\0stackql\0\0"; + let total_len = 4 + 4 + params.len(); // length field + protocol + params + + let mut msg = Vec::with_capacity(total_len); + msg.extend_from_slice(&(total_len as i32).to_be_bytes()); + msg.extend_from_slice(&PROTOCOL_V3.to_be_bytes()); + msg.extend_from_slice(params); + + self.stream + .write_all(&msg) + .map_err(|e| format!("Startup write error: {}", e))?; + + // Process auth / parameter-status messages until ReadyForQuery + loop { + let msg_type = self.read_byte()?; + let payload_len = self.read_i32()? as usize; + // payload_len includes the 4 bytes of the length field itself + let data = self.read_bytes(payload_len.saturating_sub(4))?; + + match msg_type { + b'R' => { + // AuthenticationRequest + let auth_type = + i32::from_be_bytes(data[..4].try_into().map_err(|_| "Bad auth")?); + if auth_type != 0 { + return Err(format!( + "Unsupported authentication type {} from server", + auth_type + )); + } + // AuthenticationOk — nothing to do + } + b'K' => {} // BackendKeyData — ignore + b'S' => {} // ParameterStatus — ignore + b'Z' => break, // ReadyForQuery + b'E' => return Err(parse_error_fields(&data)), + b'N' => {} // NoticeResponse during startup — ignore + _ => {} // Unknown message type — skip + } + } + + Ok(()) + } + + // ------------------------------------------------------------------ + // Query + // ------------------------------------------------------------------ + + /// Execute a simple (non-prepared) SQL query and return structured results. + pub fn query(&mut self, sql: &str) -> Result { + // Send Query message: 'Q' | int32(len) | sql\0 + let sql_bytes = sql.as_bytes(); + let payload_len = 4 + sql_bytes.len() + 1; // length field + sql + null + + let mut msg = Vec::with_capacity(1 + payload_len); + msg.push(b'Q'); + msg.extend_from_slice(&(payload_len as i32).to_be_bytes()); + msg.extend_from_slice(sql_bytes); + msg.push(0u8); + + self.stream + .write_all(&msg) + .map_err(|e| format!("Query write error: {}", e))?; + + // Collect response messages + let mut column_names: Vec = Vec::new(); + let mut rows: Vec> = Vec::new(); + let mut notices: Vec = Vec::new(); + let mut row_count: usize = 0; + + loop { + let msg_type = self.read_byte()?; + let payload_len = self.read_i32()? as usize; + let data = self.read_bytes(payload_len.saturating_sub(4))?; + + match msg_type { + b'T' => { + // RowDescription + column_names = parse_row_description(&data); + } + b'D' => { + // DataRow + let row = parse_data_row(&data, &column_names); + rows.push(row); + } + b'C' => { + // CommandComplete — tag like "SELECT 5", "INSERT 0 1", "UPDATE 3" + let tag = std::str::from_utf8(data.strip_suffix(b"\0").unwrap_or(&data)) + .unwrap_or("") + .to_string(); + if let Some(n) = tag.split_whitespace().last().and_then(|s| s.parse().ok()) { + row_count = n; + } + } + b'N' => { + notices.push(parse_notice_fields(&data)); + } + b'E' => { + return Err(parse_error_fields(&data)); + } + b'I' => {} // EmptyQueryResponse + b'Z' => break, // ReadyForQuery — done + _ => {} + } + } + + Ok(PgQueryResult { + column_names, + rows, + notices, + row_count, + }) + } + + // ------------------------------------------------------------------ + // Low-level I/O helpers + // ------------------------------------------------------------------ + + fn read_byte(&mut self) -> Result { + let mut buf = [0u8; 1]; + self.stream + .read_exact(&mut buf) + .map_err(|e| format!("Read error: {}", e))?; + Ok(buf[0]) + } + + fn read_i32(&mut self) -> Result { + let mut buf = [0u8; 4]; + self.stream + .read_exact(&mut buf) + .map_err(|e| format!("Read error: {}", e))?; + Ok(i32::from_be_bytes(buf)) + } + + fn read_bytes(&mut self, n: usize) -> Result, String> { + let mut buf = vec![0u8; n]; + self.stream + .read_exact(&mut buf) + .map_err(|e| format!("Read error: {}", e))?; + Ok(buf) + } +} + +// ------------------------------------------------------------------ +// Message parsers (free functions for readability) +// ------------------------------------------------------------------ + +fn parse_row_description(data: &[u8]) -> Vec { + let mut names = Vec::new(); + if data.len() < 2 { + return names; + } + let num_fields = u16::from_be_bytes([data[0], data[1]]) as usize; + let mut pos = 2; + + for _ in 0..num_fields { + // Null-terminated field name + let Some(null_off) = data[pos..].iter().position(|&b| b == 0) else { + break; + }; + let name = String::from_utf8_lossy(&data[pos..pos + null_off]).into_owned(); + names.push(name); + // Skip: name + null(1) + tableOID(4) + attrNum(2) + typeOID(4) + typeSize(2) + // + typeMod(4) + formatCode(2) = 19 bytes after the null + pos += null_off + 1 + 18; + } + names +} + +fn parse_data_row(data: &[u8], columns: &[String]) -> HashMap { + let mut row = HashMap::new(); + if data.len() < 2 { + return row; + } + let num_cols = u16::from_be_bytes([data[0], data[1]]) as usize; + let mut pos = 2; + + for col_name in columns.iter().take(num_cols) { + if pos + 4 > data.len() { + break; + } + let col_len = i32::from_be_bytes([data[pos], data[pos + 1], data[pos + 2], data[pos + 3]]); + pos += 4; + + let value = if col_len < 0 { + Value::Null + } else { + let len = col_len as usize; + if pos + len > data.len() { + break; + } + let s = String::from_utf8_lossy(&data[pos..pos + len]).into_owned(); + pos += len; + Value::String(s) + }; + + row.insert(col_name.clone(), value); + } + row +} + +fn parse_notice_fields(data: &[u8]) -> Notice { + let mut fields = HashMap::new(); + let mut pos = 0; + + while pos < data.len() { + let field_code = data[pos]; + pos += 1; + if field_code == 0 { + break; + } + let Some(null_off) = data[pos..].iter().position(|&b| b == 0) else { + break; + }; + let value = String::from_utf8_lossy(&data[pos..pos + null_off]).into_owned(); + pos += null_off + 1; + + let key = match field_code { + b'S' => "severity", + b'M' => "message", + b'D' => "detail", + b'H' => "hint", + b'C' => "code", + b'P' => "position", + b'W' => "where", + _ => continue, + }; + fields.insert(key.to_string(), value); + } + + Notice { fields } +} + +fn parse_error_fields(data: &[u8]) -> String { + let mut pos = 0; + while pos < data.len() { + let field_code = data[pos]; + pos += 1; + if field_code == 0 { + break; + } + let Some(null_off) = data[pos..].iter().position(|&b| b == 0) else { + break; + }; + let value = String::from_utf8_lossy(&data[pos..pos + null_off]).into_owned(); + pos += null_off + 1; + if field_code == b'M' { + return value; + } + } + "Unknown server error".to_string() +}