loopdb: add postgres store

This commit adds a postgres store to the loopdb package.
Ths postgres migrator uses a replacer filesystem to replace the
sqlite types to postgres types in the migration.

loopdb/postgres.go

@@ -0,0 +1,135 @@
+package loopdb
+
+import (
+	"database/sql"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/btcsuite/btcd/chaincfg"
+	postgres_migrate "github.com/golang-migrate/migrate/v4/database/postgres"
+	_ "github.com/golang-migrate/migrate/v4/source/file"
+	"github.com/lightninglabs/loop/loopdb/sqlc"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	dsnTemplate = "postgres://%v:%v@%v:%d/%v?sslmode=%v"
+)
+
+var (
+	// DefaultPostgresFixtureLifetime is the default maximum time a Postgres
+	// test fixture is being kept alive. After that time the docker
+	// container will be terminated forcefully, even if the tests aren't
+	// fully executed yet. So this time needs to be chosen correctly to be
+	// longer than the longest expected individual test run time.
+	DefaultPostgresFixtureLifetime = 10 * time.Minute
+)
+
+// PostgresConfig holds the postgres database configuration.
+type PostgresConfig struct {
+	SkipMigrations     bool   `long:"skipmigrations" description:"Skip applying migrations on startup."`
+	Host               string `long:"host" description:"Database server hostname."`
+	Port               int    `long:"port" description:"Database server port."`
+	User               string `long:"user" description:"Database user."`
+	Password           string `long:"password" description:"Database user's password."`
+	DBName             string `long:"dbname" description:"Database name to use."`
+	MaxOpenConnections int32  `long:"maxconnections" description:"Max open connections to keep alive to the database server."`
+	RequireSSL         bool   `long:"requiressl" description:"Whether to require using SSL (mode: require) when connecting to the server."`
+}
+
+// DSN returns the dns to connect to the database.
+func (s *PostgresConfig) DSN(hidePassword bool) string {
+	var sslMode = "disable"
+	if s.RequireSSL {
+		sslMode = "require"
+	}
+
+	password := s.Password
+	if hidePassword {
+		// Placeholder used for logging the DSN safely.
+		password = "****"
+	}
+
+	return fmt.Sprintf(dsnTemplate, s.User, password, s.Host, s.Port,
+		s.DBName, sslMode)
+}
+
+// PostgresStore is a database store implementation that uses a Postgres
+// backend.
+type PostgresStore struct {
+	cfg *PostgresConfig
+
+	*BaseDB
+}
+
+// NewPostgresStore creates a new store that is backed by a Postgres database
+// backend.
+func NewPostgresStore(cfg *PostgresConfig,
+	network *chaincfg.Params) (*PostgresStore, error) {
+
+	log.Infof("Using SQL database '%s'", cfg.DSN(true))
+
+	rawDb, err := sql.Open("pgx", cfg.DSN(false))
+	if err != nil {
+		return nil, err
+	}
+
+	if !cfg.SkipMigrations {
+		// Now that the database is open, populate the database with
+		// our set of schemas based on our embedded in-memory file
+		// system.
+		//
+		// First, we'll need to open up a new migration instance for
+		// our current target database: sqlite.
+		driver, err := postgres_migrate.WithInstance(
+			rawDb, &postgres_migrate.Config{},
+		)
+		if err != nil {
+			return nil, err
+		}
+
+		postgresFS := newReplacerFS(sqlSchemas, map[string]string{
+			"BLOB":                "BYTEA",
+			"INTEGER PRIMARY KEY": "SERIAL PRIMARY KEY",
+		})
+
+		err = applyMigrations(
+			postgresFS, driver, "sqlc/migrations", cfg.DBName,
+		)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	queries := sqlc.New(rawDb)
+
+	return &PostgresStore{
+		cfg: cfg,
+		BaseDB: &BaseDB{
+			DB:      rawDb,
+			Queries: queries,
+			network: network,
+		},
+	}, nil
+}
+
+// NewTestPostgresDB is a helper function that creates a Postgres database for
+// testing.
+func NewTestPostgresDB(t *testing.T) *PostgresStore {
+	t.Helper()
+
+	t.Logf("Creating new Postgres DB for testing")
+
+	sqlFixture := NewTestPgFixture(t, DefaultPostgresFixtureLifetime)
+	store, err := NewPostgresStore(
+		sqlFixture.GetConfig(), &chaincfg.MainNetParams,
+	)
+	require.NoError(t, err)
+
+	t.Cleanup(func() {
+		sqlFixture.TearDown(t)
+	})
+
+	return store
+}

loopdb/postgres_fixture.go

@@ -0,0 +1,139 @@
+package loopdb
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	_ "github.com/lib/pq"
+	"github.com/ory/dockertest/v3"
+	"github.com/ory/dockertest/v3/docker"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	testPgUser   = "test"
+	testPgPass   = "test"
+	testPgDBName = "test"
+	PostgresTag  = "11"
+)
+
+// TestPgFixture is a test fixture that starts a Postgres 11 instance in a
+// docker container.
+type TestPgFixture struct {
+	db       *sql.DB
+	pool     *dockertest.Pool
+	resource *dockertest.Resource
+	host     string
+	port     int
+}
+
+// NewTestPgFixture constructs a new TestPgFixture starting up a docker
+// container running Postgres 11. The started container will expire in after
+// the passed duration.
+func NewTestPgFixture(t *testing.T, expiry time.Duration) *TestPgFixture {
+	// Use a sensible default on Windows (tcp/http) and linux/osx (socket)
+	// by specifying an empty endpoint.
+	pool, err := dockertest.NewPool("")
+	require.NoError(t, err, "Could not connect to docker")
+
+	// Pulls an image, creates a container based on it and runs it.
+	resource, err := pool.RunWithOptions(&dockertest.RunOptions{
+		Repository: "postgres",
+		Tag:        PostgresTag,
+		Env: []string{
+			fmt.Sprintf("POSTGRES_USER=%v", testPgUser),
+			fmt.Sprintf("POSTGRES_PASSWORD=%v", testPgPass),
+			fmt.Sprintf("POSTGRES_DB=%v", testPgDBName),
+			"listen_addresses='*'",
+		},
+		Cmd: []string{
+			"postgres",
+			"-c", "log_statement=all",
+			"-c", "log_destination=stderr",
+		},
+	}, func(config *docker.HostConfig) {
+		// Set AutoRemove to true so that stopped container goes away
+		// by itself.
+		config.AutoRemove = true
+		config.RestartPolicy = docker.RestartPolicy{Name: "no"}
+	})
+	require.NoError(t, err, "Could not start resource")
+
+	hostAndPort := resource.GetHostPort("5432/tcp")
+	parts := strings.Split(hostAndPort, ":")
+	host := parts[0]
+	port, err := strconv.ParseInt(parts[1], 10, 64)
+	require.NoError(t, err)
+
+	fixture := &TestPgFixture{
+		host: host,
+		port: int(port),
+	}
+	databaseURL := fixture.GetDSN()
+	log.Infof("Connecting to Postgres fixture: %v\n", databaseURL)
+
+	// Tell docker to hard kill the container in "expiry" seconds.
+	require.NoError(t, resource.Expire(uint(expiry.Seconds())))
+
+	// Exponential backoff-retry, because the application in the container
+	// might not be ready to accept connections yet.
+	pool.MaxWait = 120 * time.Second
+
+	var testDB *sql.DB
+	err = pool.Retry(func() error {
+		testDB, err = sql.Open("postgres", databaseURL)
+		if err != nil {
+			return err
+		}
+		return testDB.Ping()
+	})
+	require.NoError(t, err, "Could not connect to docker")
+
+	// Now fill in the rest of the fixture.
+	fixture.db = testDB
+	fixture.pool = pool
+	fixture.resource = resource
+
+	return fixture
+}
+
+// GetDSN returns the DSN (Data Source Name) for the started Postgres node.
+func (f *TestPgFixture) GetDSN() string {
+	return f.GetConfig().DSN(false)
+}
+
+// GetConfig returns the full config of the Postgres node.
+func (f *TestPgFixture) GetConfig() *PostgresConfig {
+	return &PostgresConfig{
+		Host:       f.host,
+		Port:       f.port,
+		User:       testPgUser,
+		Password:   testPgPass,
+		DBName:     testPgDBName,
+		RequireSSL: false,
+	}
+}
+
+// TearDown stops the underlying docker container.
+func (f *TestPgFixture) TearDown(t *testing.T) {
+	err := f.pool.Purge(f.resource)
+	require.NoError(t, err, "Could not purge resource")
+}
+
+// ClearDB clears the database.
+func (f *TestPgFixture) ClearDB(t *testing.T) {
+	dbConn, err := sql.Open("postgres", f.GetDSN())
+	require.NoError(t, err)
+
+	_, err = dbConn.ExecContext(
+		context.Background(),
+		`DROP SCHEMA IF EXISTS public CASCADE;
+		 CREATE SCHEMA public;`,
+	)
+	require.NoError(t, err)
+}

loopdb/test_postgres.go

@@ -0,0 +1,13 @@
+//go:build test_db_postgres
+// +build test_db_postgres
+
+package loopdb
+
+import (
+	"testing"
+)
+
+// NewTestDB is a helper function that creates a Postgres database for testing.
+func NewTestDB(t *testing.T) *PostgresStore {
+	return NewTestPostgresDB(t)
+}

loopdb/test_sqlite.go

@@ -0,0 +1,13 @@
+//go:build !test_db_postgres
+// +build !test_db_postgres
+
+package loopdb
+
+import (
+	"testing"
+)
+
+// NewTestDB is a helper function that creates an SQLite database for testing.
+func NewTestDB(t *testing.T) *SqliteSwapStore {
+	return NewTestSqliteDB(t)
+}