1/**
2 * Regenerates the goldens in `goldens.ts` against a live PostgreSQL server.
3 *
4 * This is a maintenance tool, not a test: the unit tests only read
5 * `goldens.ts`. Run it when the fixtures need refreshing, then paste the
6 * printed JSON into `goldens.ts`.
7 *
8 * ```sh
9 * docker run -d --name pg-goldens -p 55432:5432 \
10 * -e POSTGRES_USER=effect -e POSTGRES_PASSWORD=secret -e POSTGRES_DB=effect \
11 * -e POSTGRES_HOST_AUTH_METHOD=scram-sha-256 postgres:16-alpine
12 * node packages/sql/pg/test/fixtures/regenerate.ts
13 * ```
14 *
15 * The SCRAM fixtures depend on the salt and server nonce, so each run
16 * produces a different - but equally valid - captured exchange. Capturing
17 * `authenticationMD5Password` and `authenticationCleartextPassword` needs a
18 * server started with `POSTGRES_HOST_AUTH_METHOD=md5` or `=password`.
19 */
20import { PgAuth, PgProtocol, PgTypes } from "@effect/sql-pg"
21import * as Result from "effect/Result"
22import * as net from "node:net"
23
24const success = <A, E>(result: Result.Result<A, E>): A => {
25 if (Result.isFailure(result)) throw result.failure
26 return result.success
27}
28
29const HOST = process.env.PGHOST ?? "127.0.0.1"
30const PORT = Number(process.env.PGPORT ?? 55432)
31const USER = process.env.PGUSER ?? "effect"
32const PASSWORD = process.env.PGPASSWORD ?? "secret"
33const DATABASE = process.env.PGDATABASE ?? "effect"
34const CLIENT_NONCE = "effectnonce0123456789"
35
36const toHex = (value: Uint8Array): string => Array.from(value, (byte) => byte.toString(16).padStart(2, "0")).join("")
37
38/** A connection that records the raw bytes of every backend message. */
39class Session {
40 readonly socket: net.Socket
41 readonly parser = PgProtocol.makeParser()
42 readonly frames: Array<{ readonly tag: string; readonly hex: string }> = []
43 private buffer = Buffer.alloc(0)
44 private readonly pending: Array<PgProtocol.BackendMessage> = []
45 private readonly waiters: Array<() => void> = []
46
47 constructor(socket: net.Socket) {
48 this.socket = socket
49 socket.on("data", (chunk) => this.onData(chunk as Buffer))
50 }
51
52 /** Connects, negotiates TLS away, and starts recording frames. */
53 static async connect(): Promise<{ readonly session: Session; readonly sslResponse: "S" | "N" }> {
54 const socket = net.connect({ host: HOST, port: PORT })
55 await new Promise<void>((resolve, reject) => {
56 socket.once("connect", resolve)
57 socket.once("error", reject)
58 })
59 socket.write(PgProtocol.encodeSslRequest())
60 const byte = await new Promise<number>((resolve) => socket.once("data", (data) => resolve((data as Buffer)[0])))
61 return { session: new Session(socket), sslResponse: success(PgProtocol.decodeSslResponse(byte)) }
62 }
63
64 private onData(chunk: Buffer): void {
65 this.buffer = Buffer.concat([this.buffer, chunk])
66 while (this.buffer.length >= 5) {
67 const length = this.buffer.readInt32BE(1)
68 if (this.buffer.length < length + 1) break
69 const frame = new Uint8Array(this.buffer.subarray(0, length + 1))
70 this.buffer = this.buffer.subarray(length + 1)
71 for (const message of this.parser.push(frame)) {
72 this.frames.push({ tag: message._tag, hex: toHex(frame) })
73 this.pending.push(message)
74 }
75 }
76 while (this.waiters.length > 0) this.waiters.pop()!()
77 }
78
79 async next(): Promise<PgProtocol.BackendMessage> {
80 while (this.pending.length === 0) {
81 await new Promise<void>((resolve) => this.waiters.push(resolve))
82 }
83 return this.pending.shift()!
84 }
85
86 write(bytes: Uint8Array): void {
87 this.socket.write(bytes)
88 }
89
90 /** The first frame with this tag recorded at or after `from`. */
91 frameFrom(tag: string, from: number): string | undefined {
92 return this.frames.slice(from).find((frame) => frame.tag === tag)?.hex
93 }
94
95 /** Runs an extended-query cycle and returns every message it produced. */
96 async query(
97 sql: string,
98 parameterTypes: ReadonlyArray<number> = [],
99 parameters: ReadonlyArray<Uint8Array | null> = []
100 ): Promise<Array<PgProtocol.BackendMessage>> {
101 this.write(success(PgProtocol.encodeParse({ name: "", query: sql, parameterTypes })))
102 this.write(success(PgProtocol.encodeBind({ portal: "", statement: "", parameters })))
103 this.write(PgProtocol.encodeDescribe({ target: "portal", name: "" }))
104 this.write(PgProtocol.encodeExecute({ portal: "", maxRows: 0 }))
105 this.write(PgProtocol.encodeSync())
106 const messages: Array<PgProtocol.BackendMessage> = []
107 for (;;) {
108 const message = await this.next()
109 messages.push(message)
110 if (message._tag === "ReadyForQuery") return messages
111 }
112 }
113}
114
115const authenticate = async (session: Session): Promise<void> => {
116 session.write(PgProtocol.encodeStartupMessage({
117 user: USER,
118 database: DATABASE,
119 application_name: "effect-pg-codec"
120 }))
121 let started: { readonly state: PgAuth.ScramFirst; readonly response: Uint8Array } | undefined
122 let continued: { readonly state: PgAuth.ScramFinal; readonly response: Uint8Array } | undefined
123 for (;;) {
124 const message = await session.next()
125 switch (message._tag) {
126 case "AuthenticationCleartextPassword": {
127 session.write(PgProtocol.encodePasswordMessage({ password: PASSWORD }))
128 break
129 }
130 case "AuthenticationMD5Password": {
131 const password = success(PgAuth.md5Password({ user: USER, password: PASSWORD, salt: message.salt }))
132 console.log(`md5.salt ${toHex(message.salt)}`)
133 console.log(`md5.expected ${password}`)
134 session.write(PgProtocol.encodePasswordMessage({ password }))
135 break
136 }
137 case "AuthenticationSASL": {
138 started = success(PgAuth.scramInit({ password: PASSWORD, nonce: CLIENT_NONCE }))
139 session.write(PgProtocol.encodeSASLInitialResponse({
140 mechanism: PgAuth.SCRAM_SHA_256,
141 initialResponse: started.response
142 }))
143 break
144 }
145 case "AuthenticationSASLContinue": {
146 continued = success(PgAuth.scramContinue(started!.state, message.data))
147 console.log(`scram.serverFirstMessage ${new TextDecoder().decode(message.data)}`)
148 console.log(`scram.clientFinalMessage ${new TextDecoder().decode(continued.response)}`)
149 session.write(PgProtocol.encodeSASLResponse({ data: continued.response }))
150 break
151 }
152 case "AuthenticationSASLFinal": {
153 success(PgAuth.scramFinish(continued!.state, message.data))
154 console.log(`scram.serverFinalMessage ${new TextDecoder().decode(message.data)}`)
155 break
156 }
157 case "ErrorResponse":
158 throw new Error(`Authentication failed: ${JSON.stringify(message.fields)}`)
159 case "ReadyForQuery":
160 return
161 }
162 }
163}
164
165const rowCases: ReadonlyArray<
166 { readonly name: string; readonly type: string; readonly oid: number; readonly value: unknown }
167> = [
168 { name: "bool", type: "bool", oid: PgTypes.OID.bool, value: true },
169 { name: "int2", type: "int2", oid: PgTypes.OID.int2, value: -12345 },
170 { name: "int4", type: "int4", oid: PgTypes.OID.int4, value: 2147483647 },
171 { name: "int8Max", type: "int8", oid: PgTypes.OID.int8, value: BigInt("9223372036854775807") },
172 { name: "int8Min", type: "int8", oid: PgTypes.OID.int8, value: BigInt("-9223372036854775808") },
173 { name: "oid", type: "oid", oid: PgTypes.OID.oid, value: 4294967295 },
174 { name: "regclass", type: "regclass", oid: PgTypes.OID.regclass, value: 4294967295 },
175 { name: "float4", type: "float4", oid: PgTypes.OID.float4, value: 1.5 },
176 { name: "float8", type: "float8", oid: PgTypes.OID.float8, value: -3.0625 },
177 { name: "numeric", type: "numeric", oid: PgTypes.OID.numeric, value: "12345.6789" },
178 { name: "numericSmall", type: "numeric", oid: PgTypes.OID.numeric, value: "0.0001" },
179 { name: "numericNegative", type: "numeric", oid: PgTypes.OID.numeric, value: "-98765432109876543210" },
180 { name: "numericNaN", type: "numeric", oid: PgTypes.OID.numeric, value: "NaN" },
181 { name: "text", type: "text", oid: PgTypes.OID.text, value: "héllo ☃" },
182 { name: "varchar", type: "varchar", oid: PgTypes.OID.varchar, value: "abc" },
183 { name: "bpchar", type: "bpchar", oid: PgTypes.OID.bpchar, value: "xy" },
184 { name: "name", type: "name", oid: PgTypes.OID.name, value: "some_name" },
185 { name: "bytea", type: "bytea", oid: PgTypes.OID.bytea, value: new Uint8Array([0, 1, 254, 255]) },
186 { name: "json", type: "json", oid: PgTypes.OID.json, value: { a: [1, 2], b: null } },
187 { name: "jsonb", type: "jsonb", oid: PgTypes.OID.jsonb, value: { a: [1, 2], b: null } },
188 { name: "uuid", type: "uuid", oid: PgTypes.OID.uuid, value: "6ba7b810-9dad-11d1-80b4-00c04fd430c8" },
189 { name: "inet4", type: "inet", oid: PgTypes.OID.inet, value: "192.168.0.1" },
190 { name: "inet4Masked", type: "inet", oid: PgTypes.OID.inet, value: "192.168.0.1/24" },
191 { name: "inet6", type: "inet", oid: PgTypes.OID.inet, value: "2001:db8::1" },
192 { name: "cidr", type: "cidr", oid: PgTypes.OID.cidr, value: "10.0.0.0/8" },
193 { name: "date", type: "date", oid: PgTypes.OID.date, value: "2024-02-29" },
194 { name: "dateInfinity", type: "date", oid: PgTypes.OID.date, value: "infinity" },
195 { name: "dateNegInfinity", type: "date", oid: PgTypes.OID.date, value: "-infinity" },
196 { name: "time", type: "time", oid: PgTypes.OID.time, value: BigInt(45296000000) },
197 { name: "timetz", type: "timetz", oid: PgTypes.OID.timetz, value: "12:34:56+02:00" },
198 { name: "timestamp", type: "timestamp", oid: PgTypes.OID.timestamp, value: new Date(1717171717123) },
199 { name: "timestampInfinity", type: "timestamp", oid: PgTypes.OID.timestamp, value: Number.POSITIVE_INFINITY },
200 { name: "timestamptz", type: "timestamptz", oid: PgTypes.OID.timestamptz, value: new Date(1717171717123) },
201 {
202 name: "timestamptzNegInfinity",
203 type: "timestamptz",
204 oid: PgTypes.OID.timestamptz,
205 value: Number.NEGATIVE_INFINITY
206 },
207 { name: "int4ArrayWithNulls", type: "int4[]", oid: PgTypes.OID.int4Array, value: [1, null, -3] },
208 { name: "textArrayEmpty", type: "text[]", oid: PgTypes.OID.textArray, value: [] },
209 {
210 name: "timestamptzArray",
211 type: "timestamptz[]",
212 oid: PgTypes.OID.timestamptzArray,
213 value: [new Date(0), null, new Date(1717171717000)]
214 }
215]
216
217const main = async () => {
218 const frontend: Record<string, string> = {
219 sslRequest: toHex(PgProtocol.encodeSslRequest()),
220 cancelRequest: toHex(PgProtocol.encodeCancelRequest({ pid: 63, secret: 166060928 })),
221 startupMessage: toHex(
222 PgProtocol.encodeStartupMessage({ user: USER, database: DATABASE, application_name: "effect-pg-codec" })
223 ),
224 parse: toHex(success(PgProtocol.encodeParse({ name: "s1", query: "SELECT $1", parameterTypes: [23] }))),
225 bind: toHex(
226 success(PgProtocol.encodeBind({
227 portal: "p1",
228 statement: "s1",
229 parameters: [new Uint8Array([0, 0, 0, 1]), null]
230 }))
231 ),
232 execute: toHex(PgProtocol.encodeExecute({ portal: "p1", maxRows: 5 })),
233 describeStatement: toHex(PgProtocol.encodeDescribe({ target: "statement", name: "s1" })),
234 closePortal: toHex(PgProtocol.encodeClose({ target: "portal", name: "p1" })),
235 sync: toHex(PgProtocol.encodeSync()),
236 flush: toHex(PgProtocol.encodeFlush()),
237 terminate: toHex(PgProtocol.encodeTerminate()),
238 passwordMessage: toHex(PgProtocol.encodePasswordMessage({ password: "md5abc" })),
239 saslInitialResponse: toHex(
240 PgProtocol.encodeSASLInitialResponse({
241 mechanism: PgAuth.SCRAM_SHA_256,
242 initialResponse: new Uint8Array([110, 44, 44])
243 })
244 ),
245 saslInitialResponseEmpty: toHex(
246 PgProtocol.encodeSASLInitialResponse({ mechanism: PgAuth.SCRAM_SHA_256, initialResponse: null })
247 ),
248 saslResponse: toHex(PgProtocol.encodeSASLResponse({ data: new Uint8Array([1, 2, 3]) }))
249 }
250
251 const { session, sslResponse } = await Session.connect()
252 console.log(`sslResponse ${sslResponse}`)
253 await authenticate(session)
254
255 const backend: Record<string, string> = {}
256 const rows: Record<string, string> = {}
257 for (const frame of session.frames) {
258 if (frame.tag.startsWith("Authentication")) {
259 backend[`authentication${frame.tag.slice("Authentication".length)}`] ??= frame.hex
260 }
261 if (frame.tag === "ParameterStatus") backend.parameterStatus ??= frame.hex
262 if (frame.tag === "BackendKeyData") backend.backendKeyData = frame.hex
263 if (frame.tag === "ReadyForQuery") backend.readyForQuery = frame.hex
264 }
265
266 for (const { name, oid, type, value } of rowCases) {
267 const from = session.frames.length
268 const messages = await session.query(`SELECT $1::${type}`, [oid], [success(PgTypes.encode(value, oid))])
269 const error = messages.find((message) => message._tag === "ErrorResponse")
270 if (error !== undefined) {
271 throw new Error(`${name}: ${JSON.stringify((error as PgProtocol.ErrorResponse).fields)}`)
272 }
273 rows[name] = session.frameFrom("DataRow", from)!
274 }
275
276 {
277 const from = session.frames.length
278 await session.query("SELECT 1::int4 AS a, 'x'::text AS b")
279 backend.rowDescription = session.frameFrom("RowDescription", from)!
280 backend.commandComplete = session.frameFrom("CommandComplete", from)!
281 backend.dataRowTwoColumns = session.frameFrom("DataRow", from)!
282 backend.parseComplete = session.frameFrom("ParseComplete", from)!
283 backend.bindComplete = session.frameFrom("BindComplete", from)!
284 }
285 {
286 const from = session.frames.length
287 session.write(success(PgProtocol.encodeParse({
288 name: "",
289 query: "SELECT $1::int8, NULL::text",
290 parameterTypes: [PgTypes.OID.int8]
291 })))
292 session.write(PgProtocol.encodeDescribe({ target: "statement", name: "" }))
293 session.write(success(PgProtocol.encodeBind({
294 portal: "",
295 statement: "",
296 parameters: [success(PgTypes.encode(BigInt(7), PgTypes.OID.int8))]
297 })))
298 session.write(PgProtocol.encodeExecute({ portal: "", maxRows: 0 }))
299 session.write(PgProtocol.encodeSync())
300 while ((await session.next())._tag !== "ReadyForQuery") { /* drain */ }
301 backend.parameterDescription = session.frameFrom("ParameterDescription", from)!
302 backend.dataRowWithNull = session.frameFrom("DataRow", from)!
303 }
304 {
305 const from = session.frames.length
306 await session.query("SELECT 1/0")
307 backend.errorResponse = session.frameFrom("ErrorResponse", from)!
308 }
309 {
310 const from = session.frames.length
311 await session.query("")
312 backend.emptyQueryResponse = session.frameFrom("EmptyQueryResponse", from)!
313 backend.noData = session.frameFrom("NoData", from)!
314 }
315 {
316 const from = session.frames.length
317 session.write(
318 success(PgProtocol.encodeParse({
319 name: "",
320 query: "SELECT g FROM generate_series(1,3) g",
321 parameterTypes: []
322 }))
323 )
324 session.write(success(PgProtocol.encodeBind({ portal: "", statement: "", parameters: [] })))
325 session.write(PgProtocol.encodeExecute({ portal: "", maxRows: 1 }))
326 session.write(PgProtocol.encodeSync())
327 while ((await session.next())._tag !== "ReadyForQuery") { /* drain */ }
328 backend.portalSuspended = session.frameFrom("PortalSuspended", from)!
329 }
330 {
331 const from = session.frames.length
332 session.write(success(PgProtocol.encodeParse({ name: "s1", query: "SELECT 1", parameterTypes: [] })))
333 session.write(PgProtocol.encodeClose({ target: "statement", name: "s1" }))
334 session.write(PgProtocol.encodeSync())
335 while ((await session.next())._tag !== "ReadyForQuery") { /* drain */ }
336 backend.closeComplete = session.frameFrom("CloseComplete", from)!
337 }
338 {
339 await session.query("LISTEN effect_channel")
340 const from = session.frames.length
341 await session.query("NOTIFY effect_channel, 'payload text'")
342 backend.notificationResponse = session.frameFrom("NotificationResponse", from) ?? backend.notificationResponse
343 }
344 {
345 const from = session.frames.length
346 await session.query("DROP TABLE IF EXISTS effect_missing_table")
347 backend.noticeResponse = session.frameFrom("NoticeResponse", from) ?? backend.noticeResponse
348 }
349
350 session.write(PgProtocol.encodeTerminate())
351 session.socket.end()
352 console.log(JSON.stringify({ frontend, backend, rows }, null, 2))
353}
354
355await main()