pizzasql/pkg/pgserver/pgx_test.go

package pgserver

import (
	"context"
	"fmt"
	"net"
	"reflect"
	"strings"
	"testing"
	"time"

	"github.com/jackc/pgx/v5"

	"github.com/danfragoso/pizzasql-next/pkg/storage"
	"github.com/danfragoso/pizzasql-next/pkg/testkv"
)

// connectPGX starts a wire server over net.Pipe and connects pgx to it, so the
// tests exercise the same Parse/Bind/Describe/Execute sequence a real client
// uses. pgx's default QueryExecMode is cache_statement.
func connectPGX(t *testing.T) *pgx.Conn {
	t.Helper()
	kv := testkv.New(t)
	pool := kv.Pool(4)
	t.Cleanup(func() { pool.Close() })
	dbManager := storage.NewDatabaseManager(pool, &storage.DatabaseManagerConfig{
		DefaultDatabase: "pgx_parameters",
		AutoCreate:      true,
	})

	server, client := net.Pipe()
	done := make(chan error, 1)
	conn := NewConnection(server, dbManager, true)
	go func() { done <- conn.Handle() }()
	t.Cleanup(func() {
		client.Close()
		server.Close()
		select {
		case <-done:
		case <-time.After(5 * time.Second):
			t.Error("server connection did not close")
		}
	})

	config, err := pgx.ParseConfig("postgres://tester@localhost:5432/pgx_parameters?sslmode=disable")
	if err != nil {
		t.Fatalf("parse pgx config: %v", err)
	}
	config.DialFunc = func(ctx context.Context, network, addr string) (net.Conn, error) {
		return client, nil
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	t.Cleanup(cancel)
	pgxConn, err := pgx.ConnectConfig(ctx, config)
	if err != nil {
		t.Fatalf("pgx connect: %v", err)
	}
	t.Cleanup(func() { _ = pgxConn.Close(context.Background()) })
	return pgxConn
}

func preparePGXSchema(t *testing.T, conn *pgx.Conn) {
	t.Helper()
	ctx := context.Background()
	for _, sql := range []string{
		`CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT, doc JSONB)`,
		`INSERT INTO t (id, name, doc) VALUES (1, 'alpha', '{"a": 1}')`,
		`INSERT INTO t (id, name, doc) VALUES (2, 'beta', '{"b": 2}')`,
	} {
		if _, err := conn.Exec(ctx, sql); err != nil {
			t.Fatalf("%s: %v", sql, err)
		}
	}
}

// ParameterDescription carries the inferred OID for every parameter, which is
// what pgx validates its argument list against.
func TestPGXDescribesInferredParameterTypes(t *testing.T) {
	conn := connectPGX(t)
	preparePGXSchema(t, conn)
	ctx := context.Background()
	cases := []struct {
		sql  string
		want []uint32
	}{
		{`SELECT $1::INTEGER`, []uint32{23}},
		{`SELECT $1::TEXT[]`, []uint32{1009}},
		{`SELECT $1::JSONB`, []uint32{3802}},
		{`SELECT id FROM t WHERE id = $1`, []uint32{23}},
		{`SELECT name FROM t WHERE name = $1`, []uint32{25}},
		{`SELECT id FROM t WHERE id = ANY($1)`, []uint32{1007}},
		{`SELECT doc FROM t WHERE doc = $1::JSONB`, []uint32{3802}},
		{`SELECT doc FROM t WHERE doc = $1`, []uint32{3802}},
		{`INSERT INTO t (id, name) VALUES ($1, $2)`, []uint32{23, 25}},
		{`UPDATE t SET name = $1 WHERE id = $2`, []uint32{25, 23}},
		{`SELECT $1`, []uint32{25}},
	}
	for i, tc := range cases {
		t.Run(tc.sql, func(t *testing.T) {
			sd, err := conn.Prepare(ctx, fmt.Sprintf("described_%d", i), tc.sql)
			if err != nil {
				t.Fatalf("prepare: %v", err)
			}
			if !reflect.DeepEqual(sd.ParamOIDs, tc.want) {
				t.Fatalf("ParamOIDs = %v, want %v", sd.ParamOIDs, tc.want)
			}
		})
	}
}

// The default connection mode (cache_statement) and the other describe-based
// modes bind and execute typed parameters.
func TestPGXDefaultQueryModes(t *testing.T) {
	conn := connectPGX(t)
	preparePGXSchema(t, conn)
	ctx := context.Background()
	modes := []struct {
		name string
		mode pgx.QueryExecMode
	}{
		{"cache_statement", pgx.QueryExecModeCacheStatement},
		{"cache_describe", pgx.QueryExecModeCacheDescribe},
		{"describe_exec", pgx.QueryExecModeDescribeExec},
	}
	for i, tc := range modes {
		insertID := int32(10 + i)
		t.Run(tc.name, func(t *testing.T) {
			var integer int32
			if err := conn.QueryRow(ctx, `SELECT $1::INTEGER`, tc.mode, int32(41)).Scan(&integer); err != nil {
				t.Fatalf("cast parameter: %v", err)
			}
			if integer != 41 {
				t.Fatalf("cast parameter = %d, want 41", integer)
			}

			var id int32
			if err := conn.QueryRow(ctx, `SELECT id FROM t WHERE id = $1`, tc.mode, int32(1)).Scan(&id); err != nil {
				t.Fatalf("comparison parameter: %v", err)
			}
			if id != 1 {
				t.Fatalf("comparison parameter = %d, want 1", id)
			}

			var ids []int32
			rows, err := conn.Query(ctx, `SELECT id FROM t WHERE id = ANY($1) ORDER BY id`, tc.mode, []int32{1, 2})
			if err != nil {
				t.Fatalf("ANY parameter: %v", err)
			}
			for rows.Next() {
				var value int32
				if err := rows.Scan(&value); err != nil {
					t.Fatalf("scan ANY row: %v", err)
				}
				ids = append(ids, value)
			}
			if err := rows.Err(); err != nil {
				t.Fatalf("ANY rows: %v", err)
			}
			if len(ids) != 2 || ids[0] != 1 || ids[1] != 2 {
				t.Fatalf("ANY rows = %v, want [1 2]", ids)
			}

			tag, err := conn.Exec(ctx, `INSERT INTO t (id, name) VALUES ($1, $2)`, tc.mode, insertID, "inserted")
			if err != nil {
				t.Fatalf("insert parameter: %v", err)
			}
			if tag.RowsAffected() != 1 {
				t.Fatalf("insert affected %d rows, want 1", tag.RowsAffected())
			}
			var name string
			if err := conn.QueryRow(ctx, `SELECT name FROM t WHERE id = $1`, insertID).Scan(&name); err != nil {
				t.Fatalf("read inserted row: %v", err)
			}
			if name != "inserted" {
				t.Fatalf("inserted name = %q", name)
			}

			var doc string
			if err := conn.QueryRow(ctx, `SELECT doc FROM t WHERE doc = $1::JSONB`, tc.mode, map[string]any{"a": 1}).Scan(&doc); err != nil {
				t.Fatalf("jsonb parameter: %v", err)
			}
			if doc != `{"a":1}` {
				t.Fatalf("jsonb parameter = %q, want {\"a\":1}", doc)
			}

			var untyped string
			if err := conn.QueryRow(ctx, `SELECT $1`, tc.mode, "hello").Scan(&untyped); err != nil {
				t.Fatalf("untyped parameter: %v", err)
			}
			if untyped != "hello" {
				t.Fatalf("untyped parameter = %q", untyped)
			}
		})
	}
}

// A client-supplied Parse OID wins over inference, and Bind decodes the value
// with it.
func TestPGXExplicitParameterOIDWins(t *testing.T) {
	conn := connectPGX(t)
	ctx := context.Background()
	sd, err := conn.PgConn().Prepare(ctx, "explicit_oid", `SELECT $1`, []uint32{20})
	if err != nil {
		t.Fatalf("prepare: %v", err)
	}
	if !reflect.DeepEqual(sd.ParamOIDs, []uint32{20}) {
		t.Fatalf("ParamOIDs = %v, want [20]", sd.ParamOIDs)
	}
	result := conn.PgConn().ExecPrepared(ctx, "explicit_oid", [][]byte{[]byte("42")}, []int16{0}, []int16{0}).Read()
	if result.Err != nil {
		t.Fatalf("execute: %v", result.Err)
	}
	if len(result.Rows) != 1 || len(result.Rows[0]) != 1 || string(result.Rows[0][0]) != "42" {
		t.Fatalf("explicit OID rows = %v, want [[42]]", result.Rows)
	}
}

// The described parameter count is the number of placeholders, so a client
// catches a wrong argument count before sending Bind instead of failing on an
// empty declaration.
func TestPGXParameterCountMismatch(t *testing.T) {
	conn := connectPGX(t)
	preparePGXSchema(t, conn)
	ctx := context.Background()
	var value int32
	for _, tc := range []struct {
		name string
		args []any
		want string
	}{
		{"missing_argument", nil, "expected 1 arguments, got 0"},
		{"extra_argument", []any{int32(1), int32(2)}, "expected 1 arguments, got 2"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			err := conn.QueryRow(ctx, `SELECT id FROM t WHERE id = $1`, tc.args...).Scan(&value)
			if err == nil {
				t.Fatal("query with a wrong argument count succeeded")
			}
			if !strings.Contains(err.Error(), tc.want) {
				t.Fatalf("error = %v, want %q", err, tc.want)
			}
		})
	}
}