Respect --schema flag (#26)

Make `pg-roll` respect the `--schema` flag by ensuring migrations are
run in the schema specified by the flag.

pkg/roll/execute_test.go

@@ -3,6 +3,7 @@ package roll_test
 import (
 	"context"
 	"database/sql"
+	"fmt"
 	"testing"
 	"time"
 
@@ -128,6 +129,41 @@ func TestSchemaIsDroppedAfterMigrationRollback(t *testing.T) {
 	})
 }
 
+func TestSchemaOptionIsRespected(t *testing.T) {
+	t.Parallel()
+
+	withMigratorInSchemaAndConnectionToContainer(t, "schema1", func(mig *roll.Roll, db *sql.DB) {
+		ctx := context.Background()
+		version := "1_create_table"
+
+		if err := mig.Start(ctx, &migrations.Migration{Name: version, Operations: migrations.Operations{createTableOp("table1")}}); err != nil {
+			t.Fatalf("Failed to start migration: %v", err)
+		}
+		if err := mig.Complete(ctx); err != nil {
+			t.Fatalf("Failed to complete migration: %v", err)
+		}
+
+		//
+		// Check that the table exists in the correct schema
+		//
+		var exists bool
+		err := db.QueryRow(`
+    SELECT EXISTS(
+      SELECT 1
+      FROM pg_catalog.pg_tables
+      WHERE tablename = $1
+      AND schemaname = $2
+    )`, "table1", "schema1").Scan(&exists)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if !exists {
+			t.Errorf("Expected table %q to exist in schema %q", "table1", "schema1")
+		}
+	})
+}
+
 func createTableOp(tableName string) *migrations.OpCreateTable {
 	return &migrations.OpCreateTable{
 		Name: tableName,
@@ -146,7 +182,7 @@ func createTableOp(tableName string) *migrations.OpCreateTable {
 	}
 }
 
-func withMigratorAndConnectionToContainer(t *testing.T, fn func(mig *roll.Roll, db *sql.DB)) {
+func withMigratorInSchemaAndConnectionToContainer(t *testing.T, schema string, fn func(mig *roll.Roll, db *sql.DB)) {
 	t.Helper()
 	ctx := context.Background()
 
@@ -182,7 +218,7 @@ func withMigratorAndConnectionToContainer(t *testing.T, fn func(mig *roll.Roll,
 	if err != nil {
 		t.Fatal(err)
 	}
-	mig, err := roll.New(ctx, cStr, "public", st)
+	mig, err := roll.New(ctx, cStr, schema, st)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -204,5 +240,14 @@ func withMigratorAndConnectionToContainer(t *testing.T, fn func(mig *roll.Roll,
 		}
 	})
 
+	_, err = db.ExecContext(ctx, fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS %s", schema))
+	if err != nil {
+		t.Fatal(err)
+	}
+
 	fn(mig, db)
 }
+
+func withMigratorAndConnectionToContainer(t *testing.T, fn func(mig *roll.Roll, db *sql.DB)) {
+	withMigratorInSchemaAndConnectionToContainer(t, "public", fn)
+}

pkg/roll/roll.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"database/sql"
 
+	"github.com/lib/pq"
+
 	"pg-roll/pkg/state"
 )
 
@@ -17,7 +19,14 @@ type Roll struct {
 }
 
 func New(ctx context.Context, pgURL, schema string, state *state.State) (*Roll, error) {
-	conn, err := sql.Open("postgres", pgURL)
+	dsn, err := pq.ParseURL(pgURL)
+	if err != nil {
+		dsn = pgURL
+	}
+
+	dsn += " search_path=" + schema
+
+	conn, err := sql.Open("postgres", dsn)
 	if err != nil {
 		return nil, err
 	}

pkg/state/state.go

@@ -35,7 +35,7 @@ CREATE TABLE IF NOT EXISTS %[1]s.migrations (
 CREATE UNIQUE INDEX IF NOT EXISTS only_one_active ON %[1]s.migrations (schema, name, done) WHERE done = false;
 
 -- Only first migration can exist without parent
-CREATE UNIQUE INDEX IF NOT EXISTS only_first_migration_without_parent ON %[1]s.migrations ((1)) WHERE parent IS NULL;
+CREATE UNIQUE INDEX IF NOT EXISTS only_first_migration_without_parent ON %[1]s.migrations (schema) WHERE parent IS NULL;
 
 -- History is linear
 CREATE UNIQUE INDEX IF NOT EXISTS history_is_linear ON %[1]s.migrations (schema, parent);
@@ -50,7 +50,12 @@ CREATE OR REPLACE FUNCTION %[1]s.is_active_migration_period(schemaname NAME) RET
 
 -- Get the latest version name (this is the one with child migrations)
 CREATE OR REPLACE FUNCTION %[1]s.latest_version(schemaname NAME) RETURNS text
-AS $$ SELECT p.name FROM %[1]s.migrations p WHERE NOT EXISTS (SELECT 1 FROM %[1]s.migrations c WHERE schema=schemaname AND c.parent=p.name) $$
+AS $$ 
+  SELECT p.name FROM %[1]s.migrations p 
+  WHERE NOT EXISTS (
+    SELECT 1 FROM %[1]s.migrations c WHERE schema=schemaname AND c.parent=p.name
+  ) 
+  AND schema=schemaname $$
 LANGUAGE SQL
 STABLE;