RenewCMS/test/postgres_server_test.go

package test

import (
	domainArticle "RenewCMS/internal/domain/article"
	domainInvite "RenewCMS/internal/domain/invite"
	"RenewCMS/internal/infrastructure/persistence/database"
	"RenewCMS/internal/infrastructure/persistence/models"
	"database/sql"
	"fmt"
	"net"
	"os"
	"path/filepath"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/jackc/pgx/v5/pgconn"
	_ "github.com/jackc/pgx/v5/stdlib"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"
)

const postgresTestDSNVariable = "RENEWCMS_TEST_POSTGRES_DSN"

var postgresTestDatabaseCounter atomic.Int64

type postgresTestServer struct {
	admin  *sql.DB
	config database.PostgresConfig
}

func connectPostgresTestServer(t *testing.T) *postgresTestServer {
	t.Helper()
	dsn := os.Getenv(postgresTestDSNVariable)
	if dsn == "" {
		t.Skip(postgresTestDSNVariable + " is not set; run task test:postgres to test against a real server")
	}
	parsed, err := pgconn.ParseConfig(dsn)
	if err != nil {
		t.Fatalf("parse %s: %v", postgresTestDSNVariable, err)
	}
	admin, err := sql.Open("pgx", dsn)
	if err != nil {
		t.Fatalf("open admin: %v", err)
	}
	t.Cleanup(func() { _ = admin.Close() })
	return &postgresTestServer{admin: admin, config: database.PostgresConfig{
		Host: parsed.Host, Port: fmt.Sprint(parsed.Port), User: parsed.User, Password: parsed.Password, SSLMode: "disable",
	}}
}

func (s *postgresTestServer) createDatabase(t *testing.T) database.PostgresConfig {
	t.Helper()
	name := fmt.Sprintf("renewcms_test_%d_%d", time.Now().UnixNano()%1_000_000_000, postgresTestDatabaseCounter.Add(1))
	if _, err := s.admin.Exec("CREATE DATABASE " + name); err != nil {
		t.Fatalf("create database: %v", err)
	}
	t.Cleanup(func() { _, _ = s.admin.Exec("DROP DATABASE IF EXISTS " + name + " WITH (FORCE)") })
	config := s.config
	config.Name = name
	return config
}

func (s *postgresTestServer) connect(t *testing.T, config database.PostgresConfig) *database.Connection {
	t.Helper()
	connection, err := database.Connect(database.Config{Driver: database.DriverPostgres, Source: database.PostgresDSN(config)})
	if err != nil {
		t.Fatalf("connect postgres: %v", err)
	}
	connection.DB.Logger = logger.Discard
	t.Cleanup(func() { _ = connection.Close() })
	return connection
}

func openContractPostgres(t *testing.T) *gorm.DB {
	t.Helper()
	server := connectPostgresTestServer(t)
	connection := server.connect(t, server.createDatabase(t))
	if err := database.Migrate(connection.DB, migrationPublicURL); err != nil {
		t.Fatalf("migrate postgres: %v", err)
	}
	return connection.DB
}

func migratedPostgres(t *testing.T) (*postgresTestServer, database.PostgresConfig, *gorm.DB) {
	t.Helper()
	server := connectPostgresTestServer(t)
	config := server.createDatabase(t)
	connection := server.connect(t, config)
	if err := database.Migrate(connection.DB, migrationPublicURL); err != nil {
		t.Fatalf("migrate: %v", err)
	}
	return server, config, connection.DB
}

func TestPostgresRepositoryContract(t *testing.T) {
	runRepositoryContract(t, contractBackend{name: "postgres", open: openContractPostgres})
}

func TestPostgresConnectionPool(t *testing.T) {
	t.Run("Connection pool", func(t *testing.T) {
		server := connectPostgresTestServer(t)
		pool := database.PoolSettings{MaxOpenConns: 4, MaxIdleConns: 2, ConnMaxLifetime: 60 * time.Second}
		connection, err := database.Connect(database.Config{Driver: database.DriverPostgres, Source: database.PostgresDSN(server.createDatabase(t)), Pool: &pool})
		if err != nil {
			t.Fatalf("connect: %v", err)
		}
		defer connection.Close()
		sqlDb, _ := connection.DB.DB()

		stats := sqlDb.Stats()

		if stats.MaxOpenConnections != 4 {
			t.Errorf("expected 4 open connections at most, got %d", stats.MaxOpenConnections)
		}
	})
}

func TestPostgresServerFaults(t *testing.T) {
	t.Run("the password is wrong", func(t *testing.T) {
		server := connectPostgresTestServer(t)
		config := server.createDatabase(t)
		config.Password = "definitely/wrong p@ss"

		_, err := database.Connect(database.Config{Driver: database.DriverPostgres, Source: database.PostgresDSN(config)})

		expectedAddress := net.JoinHostPort(config.Host, config.Port)
		if err == nil || !strings.Contains(err.Error(), expectedAddress) || strings.Contains(err.Error(), config.Password) {
			t.Errorf("unexpected error %v", err)
		}
	})

	t.Run("the database does not exist", func(t *testing.T) {
		server := connectPostgresTestServer(t)
		config := server.config
		config.Name = "renewcms_missing_database"

		_, err := database.Connect(database.Config{Driver: database.DriverPostgres, Source: database.PostgresDSN(config)})

		if err == nil || !strings.Contains(err.Error(), net.JoinHostPort(config.Host, config.Port)) || strings.Contains(err.Error(), config.Password) {
			t.Errorf("unexpected error %v", err)
		}
	})
}

func TestPostgresMigration(t *testing.T) {
	t.Run("Migration on PostgreSQL", func(t *testing.T) {
		_, _, db := migratedPostgres(t)

		for _, table := range []string{"articles", "images", "users", "invites", "api_keys", "purge_settings"} {
			count := columnCount(t, db, "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'public' AND table_name = ?", table)
			if count != 1 {
				t.Errorf("table %s must exist in schema public", table)
			}
		}
		uniqueColumns := map[string][]string{
			"users":    {"username", "email", "password_reset_code", "verification_code"},
			"api_keys": {"hashed_key"},
		}
		for table, columns := range uniqueColumns {
			for _, column := range columns {
				count := columnCount(t, db, `SELECT COUNT(*) FROM pg_index i
					JOIN pg_class c ON c.oid = i.indrelid
					JOIN pg_attribute a ON a.attrelid = c.oid AND a.attnum = i.indkey[0]
					WHERE c.relname = ? AND a.attname = ? AND i.indisunique AND array_length(i.indkey, 1) = 1`, table, column)
				if count == 0 {
					t.Errorf("a unique index on %s.%s must exist", table, column)
				}
			}
		}
	})

	t.Run("migrating twice changes nothing", func(t *testing.T) {
		_, _, db := migratedPostgres(t)
		before := columnCount(t, db, "SELECT COUNT(*) FROM pg_indexes WHERE schemaname = 'public'")

		if err := database.Migrate(db, migrationPublicURL); err != nil {
			t.Errorf("the second migration must succeed: %v", err)
		}

		if after := columnCount(t, db, "SELECT COUNT(*) FROM pg_indexes WHERE schemaname = 'public'"); after != before {
			t.Errorf("indexes changed from %d to %d", before, after)
		}
	})

	t.Run("the first user lacking a role becomes owner and the others admin", func(t *testing.T) {
		_, _, db := migratedPostgres(t)
		for _, name := range []string{"first", "second", "third"} {
			insert := db.Exec("INSERT INTO users (username, password, email, role, is_verified, token_version) VALUES (?, 'x', ?, '', true, 0)", name, name+"@x.example")
			if insert.Error != nil {
				t.Fatalf("insert %s: %v", name, insert.Error)
			}
		}

		if err := database.Migrate(db, migrationPublicURL); err != nil {
			t.Fatalf("migrate: %v", err)
		}

		roles := map[string]string{}
		var users []models.User
		db.Find(&users)
		for _, user := range users {
			roles[user.Username] = user.Role
		}
		if roles["first"] != "owner" || roles["second"] != "admin" || roles["third"] != "admin" {
			t.Errorf("unexpected roles %v", roles)
		}
	})
}

func TestPostgresTimes(t *testing.T) {
	t.Run("times are stored and read back in UTC", func(t *testing.T) {
		_, _, db := migratedPostgres(t)
		repos := newContractRepositories(db)
		zone := time.FixedZone("plus2", 2*3600)
		expiration := time.Date(2030, 5, 6, 7, 8, 9, 0, zone)
		user := contractUser("zoned", "zoned@x.example")
		user.VerificationExpiration = expiration

		created := mustCreateUser(t, repos, user)
		read, err := repos.users.Get(created.ID)

		if err != nil || !read.VerificationExpiration.Equal(expiration) || read.VerificationExpiration.Location() != time.UTC {
			t.Errorf("expected %v in UTC, got %v (%v) err %v", expiration, read.VerificationExpiration, read.VerificationExpiration.Location(), err)
		}
	})
}

func TestPostgresTimesInLists(t *testing.T) {
	_, _, db := migratedPostgres(t)
	repos := newContractRepositories(db)
	zone := time.FixedZone("plus2", 2*3600)
	for _, name := range []string{"first", "second"} {
		user := contractUser(name, name+"@x.example")
		user.VerificationExpiration = time.Date(2030, 5, 6, 7, 8, 9, 0, zone)
		mustCreateUser(t, repos, user)
	}

	users := repos.users.GetAll()

	if len(users) != 2 {
		t.Fatalf("expected 2 users, got %d", len(users))
	}
	for _, user := range users {
		if user.VerificationExpiration.Location() != time.UTC {
			t.Errorf("%s: expected UTC, got %v", user.Username, user.VerificationExpiration.Location())
		}
	}
}

func TestPostgresBackupOfIdentityAndBinaryColumns(t *testing.T) {
	server, _, db := migratedPostgres(t)
	if err := db.Exec(`CREATE TABLE extras (id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, payload bytea, label text DEFAULT 'x' NOT NULL, amount numeric(10,2))`).Error; err != nil {
		t.Fatalf("create extras: %v", err)
	}
	if err := db.Exec(`INSERT INTO extras (payload, amount) VALUES (decode('00ff27', 'hex'), 12.50), (NULL, NULL)`).Error; err != nil {
		t.Fatalf("insert extras: %v", err)
	}
	path, err := database.NewPostgresBackup(db, filepath.Join(t.TempDir(), "backups"), time.Now).Backup()
	if err != nil {
		t.Fatalf("backup: %v", err)
	}
	content, _ := os.ReadFile(path)
	dump := string(content)

	for _, fragment := range []string{
		`"id" integer GENERATED BY DEFAULT AS IDENTITY NOT NULL`,
		`"label" text DEFAULT 'x'::text NOT NULL`,
		`'\x00ff27'`,
		`12.50`,
		`NULL`,
	} {
		if !strings.Contains(dump, fragment) {
			t.Errorf("the dump must contain %s", fragment)
		}
	}

	restoredConfig := server.createDatabase(t)
	replayPostgresDump(t, restoredConfig, dump)
	restored := server.connect(t, restoredConfig)
	var payload []byte
	if err := restored.DB.Raw(`SELECT payload FROM extras WHERE id = 1`).Row().Scan(&payload); err != nil || len(payload) != 3 || payload[1] != 0xff {
		t.Errorf("unexpected restored payload %v err %v", payload, err)
	}
	var next int
	if err := restored.DB.Raw(`INSERT INTO extras (label) VALUES ('y') RETURNING id`).Row().Scan(&next); err != nil || next != 3 {
		t.Errorf("the identity sequence must continue after the restored rows, got %d err %v", next, err)
	}
}

func populateForPostgresBackup(t *testing.T, repos contractRepositories) {
	t.Helper()
	trickyName := "o'brien \"quoted\" back\\slash\nline é😀"
	mustCreateUser(t, repos, contractUser(trickyName, "tricky@x.example"))
	second := contractUser("plain", "plain@x.example")
	second.PasswordResetCode = "reset-code"
	mustCreateUser(t, repos, second)
	if _, err := repos.articles.Create(domainArticle.Article{Title: "Title", Body: "line1\nline2 'q' \"d\" \\ end", IsOnline: true}); err != nil {
		t.Fatalf("create article: %v", err)
	}
	mustCreateImage(t, repos, "pic.png", strings.Repeat("c", 64))
	if _, err := repos.invites.Create(domainInvite.Invite{Email: "i@x.example", HashedCode: "h", ExpiresAt: time.Date(2030, 1, 1, 0, 0, 0, 0, time.UTC), InvitedBy: 1}); err != nil {
		t.Fatalf("create invite: %v", err)
	}
}

func TestPostgresBackup(t *testing.T) {
	t.Run("Backup before an update", func(t *testing.T) {
		_, _, db := migratedPostgres(t)
		populateForPostgresBackup(t, newContractRepositories(db))
		directory := filepath.Join(t.TempDir(), "backups")
		backup := database.NewPostgresBackup(db, directory, steppingClock(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)))

		var paths []string
		for range 5 {
			path, err := backup.Backup()
			if err != nil {
				t.Fatalf("backup: %v", err)
			}
			paths = append(paths, path)
		}

		entries, _ := os.ReadDir(directory)
		if len(entries) != 3 {
			t.Fatalf("only the 3 newest backups are kept, got %d", len(entries))
		}
		for _, path := range paths[2:] {
			if filepath.Ext(path) != ".sql" || !strings.HasPrefix(filepath.Base(path), "renewcms-") {
				t.Errorf("unexpected backup name %q", path)
			}
			if _, err := os.Stat(path); err != nil {
				t.Errorf("the newest backup %q must remain: %v", path, err)
			}
		}
	})

	t.Run("a failure returns an error and leaves no file", func(t *testing.T) {
		server := connectPostgresTestServer(t)
		connection := server.connect(t, server.createDatabase(t))
		directory := filepath.Join(t.TempDir(), "backups")
		backup := database.NewPostgresBackup(connection.DB, directory, time.Now)
		if err := connection.Close(); err != nil {
			t.Fatalf("close: %v", err)
		}

		path, err := backup.Backup()

		if err == nil || path != "" {
			t.Errorf("expected a failure, got %q %v", path, err)
		}
		if entries, _ := os.ReadDir(directory); len(entries) != 0 {
			t.Errorf("no file may remain, got %d entries", len(entries))
		}
	})

	t.Run("PostgreSQL backup content", func(t *testing.T) {
		server, _, db := migratedPostgres(t)
		populateForPostgresBackup(t, newContractRepositories(db))
		directory := filepath.Join(t.TempDir(), "backups")

		path, err := database.NewPostgresBackup(db, directory, time.Now).Backup()
		if err != nil {
			t.Fatalf("backup: %v", err)
		}
		content, err := os.ReadFile(path)
		if err != nil {
			t.Fatal(err)
		}
		dump := string(content)

		for _, table := range []string{"articles", "images", "users", "invites", "api_keys", "purge_settings"} {
			if !strings.Contains(dump, `CREATE TABLE "`+table+`"`) {
				t.Errorf("the dump must contain the schema of %s", table)
			}
		}
		if !strings.Contains(dump, "CREATE UNIQUE INDEX") {
			t.Errorf("the dump must contain the indexes")
		}
		for _, table := range []string{"articles", "images", "users", "invites"} {
			if !strings.Contains(dump, `INSERT INTO "`+table+`"`) {
				t.Errorf("the dump must contain rows of %s", table)
			}
		}
		if password := server.config.Password; password != "" && strings.Contains(dump, password) {
			t.Errorf("the dump must not contain the DB_PASSWORD value")
		}

		restoredConfig := server.createDatabase(t)
		replayPostgresDump(t, restoredConfig, dump)
		restored := server.connect(t, restoredConfig)
		restoredPath, err := database.NewPostgresBackup(restored.DB, filepath.Join(t.TempDir(), "again"), time.Now).Backup()
		if err != nil {
			t.Fatalf("backup of the restored database: %v", err)
		}
		restoredContent, _ := os.ReadFile(restoredPath)
		if line := firstDifferingLine(dump, string(restoredContent)); line != "" {
			t.Errorf("replaying the dump must restore the same schema and rows, first difference: %s", line)
		}
		restoredRepos := newContractRepositories(restored.DB)
		restoredUsers := restoredRepos.users.GetAll()
		if len(restoredUsers) != 2 || restoredUsers[0].Username != "o'brien \"quoted\" back\\slash\nline é😀" || restoredUsers[1].PasswordResetCode != "reset-code" {
			t.Errorf("unexpected restored users %+v", restoredUsers)
		}
		restoredArticle, _ := restoredRepos.articles.Get(1)
		if restoredArticle.Body != "line1\nline2 'q' \"d\" \\ end" {
			t.Errorf("unexpected restored article body %q", restoredArticle.Body)
		}
		next := mustCreateUser(t, restoredRepos, contractUser("third", "third@x.example"))
		if next.ID != 3 {
			t.Errorf("sequences must be reset so the next insert gets an unused id, got %d", next.ID)
		}
	})
}

func replayPostgresDump(t *testing.T, config database.PostgresConfig, dump string) {
	t.Helper()
	replay, err := sql.Open("pgx", database.PostgresDSN(config))
	if err != nil {
		t.Fatal(err)
	}
	defer replay.Close()
	if _, err := replay.Exec(dump); err != nil {
		t.Fatalf("replay: %v", err)
	}
}

func TestPostgresUsersAreCaseSensitiveThroughTheDomain(t *testing.T) {
	t.Run("Bob and bob can both register", func(t *testing.T) {
		_, _, db := migratedPostgres(t)
		repos := newContractRepositories(db)
		mustCreateUser(t, repos, contractUser("Bob", "b1@x.example"))

		other, err := repos.users.Create(contractUser("bob", "b2@x.example"))
		_, missing := repos.users.GetByUsername("BOB")

		if err != nil || other.Username != "bob" || missing == nil {
			t.Errorf("created %+v err %v, uppercase lookup err %v", other, err, missing)
		}
		if _, err := repos.users.GetByEmail("B1@X.EXAMPLE"); err == nil {
			t.Error("emails are compared exactly like SQLite does")
		}
	})
}