Allow to migrate existing databases (#112)

When applying a first migration against a schema, return the schema as
read form postgres, instead of an empty one.

This change allows migrations to happen against schemas created before
pg-roll install.

pkg/state/state.go

@@ -315,12 +315,14 @@ func (s *State) Start(ctx context.Context, schemaname string, migration *migrati
 		return nil, fmt.Errorf("unable to marshal migration: %w", err)
 	}
 
+	// create a new migration object and return the previous known schema
+	// if there is no previous migration, read the schema from postgres
 	stmt := fmt.Sprintf(`
 		INSERT INTO %[1]s.migrations (schema, name, parent, migration) VALUES ($1, $2, %[1]s.latest_version($1), $3)
 		RETURNING (
 			SELECT COALESCE(
 				(SELECT resulting_schema FROM %[1]s.migrations WHERE schema=$1 AND name=%[1]s.latest_version($1)),
-				'{}')
+				%[1]s.read_schema($1))
 		)`, pq.QuoteIdentifier(s.schema))
 
 	var rawSchema string

pkg/state/state_test.go

@@ -0,0 +1,108 @@
+package state_test
+
+import (
+	"context"
+	"database/sql"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/testcontainers/testcontainers-go"
+	"github.com/testcontainers/testcontainers-go/modules/postgres"
+	"github.com/testcontainers/testcontainers-go/wait"
+	"github.com/xataio/pg-roll/pkg/migrations"
+	"github.com/xataio/pg-roll/pkg/state"
+)
+
+// The version of postgres against which the tests are run
+// if the POSTGRES_VERSION environment variable is not set.
+const defaultPostgresVersion = "15.3"
+
+func TestSchemaOptionIsRespected(t *testing.T) {
+	t.Parallel()
+
+	witStateAndConnectionToContainer(t, func(state *state.State, db *sql.DB) {
+		ctx := context.Background()
+
+		// create a table in the public schema
+		if _, err := db.ExecContext(ctx, "CREATE TABLE public.table1 (id int)"); err != nil {
+			t.Fatal(err)
+		}
+
+		// init the state
+		if err := state.Init(ctx); err != nil {
+			t.Fatal(err)
+		}
+
+		// check that starting a new migration returns the already existing table
+		currentSchema, err := state.Start(ctx, "public", &migrations.Migration{
+			Name: "1_add_column",
+			Operations: migrations.Operations{
+				&migrations.OpAddColumn{
+					Table: "table1",
+					Column: migrations.Column{
+						Name: "test",
+						Type: "text",
+					},
+				},
+			},
+		})
+		assert.NoError(t, err)
+
+		assert.Equal(t, 1, len(currentSchema.Tables))
+		assert.Equal(t, "public", currentSchema.Name)
+	})
+}
+
+func witStateAndConnectionToContainer(t *testing.T, fn func(*state.State, *sql.DB)) {
+	t.Helper()
+	ctx := context.Background()
+
+	waitForLogs := wait.
+		ForLog("database system is ready to accept connections").
+		WithOccurrence(2).
+		WithStartupTimeout(5 * time.Second)
+
+	pgVersion := os.Getenv("POSTGRES_VERSION")
+	if pgVersion == "" {
+		pgVersion = defaultPostgresVersion
+	}
+
+	ctr, err := postgres.RunContainer(ctx,
+		testcontainers.WithImage("postgres:"+pgVersion),
+		testcontainers.WithWaitStrategy(waitForLogs),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	t.Cleanup(func() {
+		if err := ctr.Terminate(ctx); err != nil {
+			t.Fatalf("Failed to terminate container: %v", err)
+		}
+	})
+
+	cStr, err := ctr.ConnectionString(ctx, "sslmode=disable")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	db, err := sql.Open("postgres", cStr)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	t.Cleanup(func() {
+		if err := db.Close(); err != nil {
+			t.Fatalf("Failed to close database connection: %v", err)
+		}
+	})
+
+	st, err := state.New(ctx, cStr, "pgroll")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	fn(st, db)
+}