member-console/internal/testkit/dbtest/open.go

// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
// SPDX-FileCopyrightText: 2025-2026 Christian Galo

// Package dbtest opens the databases the tests use and prepares scratch
// databases for the tests whose subject is a migration. It imports only
// internal/db and internal/testkit, so every package's tests can import it
// without a cycle (internal/migrate reaches nearly every module package).
//
// Open never migrates: test/reset-test-db.sh is the only migrator of the
// shared test database, and Open fails when the reset has not run.
package dbtest

import (
	"context"
	"database/sql"
	"fmt"
	"io/fs"
	"os"
	"regexp"
	"strconv"
	"strings"
	"testing"

	_ "github.com/jackc/pgx/v5/stdlib"

	"git.coopcloud.tech/wiki-cafe/member-console/internal/db"
	"git.coopcloud.tech/wiki-cafe/member-console/internal/testkit"
)

// behind is what a test sees when the shared database is not at the schema
// the code under test expects.
const behind = "the test database is behind; run test/reset-test-db.sh"

// Option changes how Open reaches the database.
type Option func(*opener)

type opener struct{ searchPath []string }

// WithSearchPath appends search_path=<schemas> to the address when it names
// none, as the app's own connection does.
func WithSearchPath(schemas ...string) Option {
	return func(o *opener) { o.searchPath = schemas }
}

// DSN returns TEST_DATABASE_URL and skips the test when it is unset.
func DSN(tb testing.TB) string {
	tb.Helper()
	dsn := os.Getenv("TEST_DATABASE_URL")
	if dsn == "" {
		tb.Skip("TEST_DATABASE_URL not set, skipping integration test")
	}
	return dsn
}

// Open opens TEST_DATABASE_URL, which test/reset-test-db.sh has migrated, and
// closes it when the test ends. It skips the test when the variable is unset
// and fails it when the core ledger lacks the highest version of
// db.BaseSources(). It never migrates.
func Open(tb testing.TB, opts ...Option) *sql.DB {
	tb.Helper()
	var o opener
	for _, opt := range opts {
		opt(&o)
	}
	dsn := DSN(tb)
	if len(o.searchPath) > 0 && !strings.Contains(dsn, "search_path") {
		dsn += "&search_path=" + strings.Join(o.searchPath, ",")
	}
	return opened(tb, dsn)
}

// OpenE2E opens E2E_DATABASE_URL, the database test/e2e/plan-management
// commits its fixtures to, as Open opens the test database.
func OpenE2E(tb testing.TB) *sql.DB {
	tb.Helper()
	dsn := os.Getenv("E2E_DATABASE_URL")
	if dsn == "" {
		tb.Skip("E2E_DATABASE_URL not set, skipping e2e test")
	}
	return opened(tb, dsn)
}

// opened connects to dsn and fails the test unless the database is migrated.
func opened(tb testing.TB, dsn string) *sql.DB {
	tb.Helper()
	database := Connect(tb, dsn)
	if err := ledgerBehind(database); err != nil {
		tb.Fatalf("%v", err)
	}
	return database
}

// Connect opens dsn with the pgx driver and closes it when the test ends.
func Connect(tb testing.TB, dsn string) *sql.DB {
	tb.Helper()
	database, err := sql.Open("pgx", dsn)
	if err != nil {
		tb.Fatalf("open database: %v", err)
	}
	tb.Cleanup(func() { testkit.Close(tb, database) })
	return database
}

// Closed returns a database that is already closed, so every call on it fails
// with sql.ErrConnDone. It reaches no server.
func Closed(tb testing.TB) *sql.DB {
	tb.Helper()
	database, err := sql.Open("pgx", "postgres://closed-database-seam/testdb")
	if err != nil {
		tb.Fatalf("open database: %v", err)
	}
	if err := database.Close(); err != nil {
		tb.Fatalf("close database: %v", err)
	}
	return database
}

var versionPrefix = regexp.MustCompile(`^(\d+)_`)

// ledgerTable is the goose version table of one migration stream.
func ledgerTable(src db.MigrationSource) string { return "goose_db_version_" + src.Name }

// highestVersion is the highest version among the stream's migration files.
func highestVersion(src db.MigrationSource) (int64, error) {
	entries, err := fs.ReadDir(src.Migrations, src.Dir)
	if err != nil {
		return 0, fmt.Errorf("read %s migrations: %w", src.Name, err)
	}
	var highest int64
	for _, e := range entries {
		m := versionPrefix.FindStringSubmatch(e.Name())
		if m == nil {
			continue
		}
		v, err := strconv.ParseInt(m[1], 10, 64)
		if err != nil {
			return 0, fmt.Errorf("migration %s: %w", e.Name(), err)
		}
		highest = max(highest, v)
	}
	return highest, nil
}

// ledgerBehind returns an error naming the reset script when the core ledger
// of database lacks the highest version in db.BaseSources().
func ledgerBehind(database *sql.DB) error {
	core := db.BaseSources()[0]
	want, err := highestVersion(core)
	if err != nil {
		return err
	}
	var applied sql.NullInt64
	err = database.QueryRowContext(context.Background(),
		`SELECT max(version_id) FROM `+ledgerTable(core)+` WHERE is_applied`).Scan(&applied)
	if err != nil {
		return fmt.Errorf("%s (core ledger: %v)", behind, err)
	}
	if applied.Int64 < want {
		return fmt.Errorf("%s (core is at version %d, want %d)", behind, applied.Int64, want)
	}
	return nil
}