refactor(schema): Add support for migration editions

This refactors the schema package and subpackages to support editions.
An Edition is a collection of sql migrations. A given boundary binary
can have multiple migration editions. The `schema.Manager` is updated to
apply the necessary migrations in the correct order, leveraging an
Edition's priority to determine the order in which the separate Editions
are applied.

The sql migration files are now nested under a directory for the
edition.

This also switches from the generated code to using embed, which
eliminates the need for the `make migrations` target and the generated
code. As such this commit removes the target and generated code.

See:
    https://pkg.go.dev/embed

Makefile

@@ -77,10 +77,7 @@ endif
 perms-table:
 	@go run internal/website/permstable/permstable.go
 
-gen: cleangen proto api cli migrations perms-table fmt
-
-migrations:
-	$(MAKE) --environment-overrides -C internal/db/schema/migrations/generate migrations
+gen: cleangen proto api cli perms-table fmt
 
 ### oplog requires protoc-gen-go v1.20.0 or later
 # GO111MODULE=on go get -u github.com/golang/protobuf/protoc-gen-go@v1.40
@@ -221,7 +218,7 @@ docker-publish:
 	docker push $(IMAGE_TAG)
 	docker push hashicorp/boundary:latest
 
-.PHONY: api cli tools gen migrations proto website ci-config ci-verify set-ui-version docker docker-build docker-build-dev docker-publish test-database-up test-database-down
+.PHONY: api cli tools gen proto website ci-config ci-verify set-ui-version docker docker-build docker-build-dev docker-publish test-database-up test-database-down
 
 .NOTPARALLEL:
 

internal/cmd/base/dev.go

@@ -65,7 +65,7 @@ func (b *Server) CreateDevDatabase(ctx context.Context, opt ...Option) error {
 		}
 
 		// Let migrate store manage the dirty bit since dev DBs should be ephemeral anyways.
-		_, err := schema.MigrateStore(ctx, dialect, url)
+		_, err := schema.MigrateStore(ctx, schema.Dialect(dialect), url)
 		if err != nil {
 			err = fmt.Errorf("unable to initialize dev database with dialect %s: %w", dialect, err)
 			if c != nil {
@@ -78,7 +78,7 @@ func (b *Server) CreateDevDatabase(ctx context.Context, opt ...Option) error {
 		b.DatabaseUrl = url
 	default:
 		// Let migrate store manage the dirty bit since dev DBs should be ephemeral anyways.
-		if _, err := schema.MigrateStore(ctx, dialect, b.DatabaseUrl); err != nil {
+		if _, err := schema.MigrateStore(ctx, schema.Dialect(dialect), b.DatabaseUrl); err != nil {
 			err = fmt.Errorf("error initializing store: %w", err)
 			if c != nil {
 				err = multierror.Append(err, c())

internal/cmd/commands/database/funcs.go

@@ -29,7 +29,7 @@ func migrateDatabase(ctx context.Context, ui cli.Ui, dialect, u string, initiali
 		ui.Error(fmt.Sprintf("Unable to connect to the database at %q", u))
 		return noop, 2
 	}
-	man, err := schema.NewManager(ctx, dialect, dBase)
+	man, err := schema.NewManager(ctx, schema.Dialect(dialect), dBase)
 	if err != nil {
 		if errors.Match(errors.T(errors.MigrationLock), err) {
 			ui.Error("Unable to capture a lock on the database.")
@@ -53,26 +53,22 @@ func migrateDatabase(ctx context.Context, ui cli.Ui, dialect, u string, initiali
 		ui.Error(fmt.Errorf("Error getting database state: %w", err).Error())
 		return unlock, 2
 	}
-	if initialized && !st.InitializationStarted {
+	if initialized && !st.Initialized {
 		ui.Output(base.WrapAtLength("Database has not been initialized. Please use 'boundary database init' to initialize the boundary database."))
 		return unlock, -1
 	}
-	if !initialized && st.InitializationStarted {
+	if !initialized && st.Initialized {
 		ui.Output(base.WrapAtLength("Database has already been initialized. Please use 'boundary database migrate' for any upgrade needs."))
 		return unlock, -1
 	}
-	if st.Dirty {
-		ui.Error(base.WrapAtLength("Database is in a bad state.  Please revert back to the last known good state."))
-		return unlock, 2
-	}
-	if err := man.RollForward(ctx); err != nil {
+	if err := man.ApplyMigrations(ctx); err != nil {
 		ui.Error(fmt.Errorf("Error running database migrations: %w", err).Error())
 		return unlock, 2
 	}
 	if base.Format(ui) == "table" {
 		ui.Info("Migrations successfully run.")
 	}
-	migrationLogs, err := schema.GetMigrationLog(ctx, dBase)
+	migrationLogs, err := man.GetMigrationLog(ctx)
 	if err != nil {
 		ui.Error(fmt.Errorf("Error retrieving database migration logs: %w", err).Error())
 		return unlock, 2

internal/cmd/commands/database/funcs_test.go

@@ -53,13 +53,11 @@ func TestMigrateDatabase(t *testing.T) {
 				require.NoError(t, err)
 
 				earlyMigrationVersion := 2000
-				oState := schema.TestCloneMigrationStates(t)
-				nState := schema.TestCreatePartialMigrationState(oState["postgres"], earlyMigrationVersion)
-				oState["postgres"] = nState
-
-				man, err := schema.NewManager(ctx, dialect, dBase, schema.WithMigrationStates(oState))
+				man, err := schema.NewManager(ctx, schema.Dialect(dialect), dBase, schema.WithEditions(
+					schema.TestCreatePartialEditions(schema.Dialect(dialect), schema.PartialEditions{"oss": earlyMigrationVersion}),
+				))
 				require.NoError(t, err)
-				require.NoError(t, man.RollForward(ctx))
+				require.NoError(t, man.ApplyMigrations(ctx))
 				return u
 			},
 			expectedCode:   0,
@@ -79,13 +77,11 @@ func TestMigrateDatabase(t *testing.T) {
 				require.NoError(t, err)
 
 				earlyMigrationVersion := 2000
-				oState := schema.TestCloneMigrationStates(t)
-				nState := schema.TestCreatePartialMigrationState(oState["postgres"], earlyMigrationVersion)
-				oState["postgres"] = nState
-
-				man, err := schema.NewManager(ctx, dialect, dBase, schema.WithMigrationStates(oState))
+				man, err := schema.NewManager(ctx, schema.Dialect(dialect), dBase, schema.WithEditions(
+					schema.TestCreatePartialEditions(schema.Dialect(dialect), schema.PartialEditions{"oss": earlyMigrationVersion}),
+				))
 				require.NoError(t, err)
-				require.NoError(t, man.RollForward(ctx))
+				require.NoError(t, man.ApplyMigrations(ctx))
 				return u
 			},
 			expectedCode:   -1,
@@ -157,7 +153,7 @@ func TestMigrateDatabase(t *testing.T) {
 				dBase, err := common.SqlOpen(dialect, u)
 				require.NoError(t, err)
 
-				man, err := schema.NewManager(ctx, dialect, dBase)
+				man, err := schema.NewManager(ctx, schema.Dialect(dialect), dBase)
 				require.NoError(t, err)
 				// This is an advisory lock on the DB which is released when the DB session ends.
 				err = man.ExclusiveLock(ctx)
@@ -180,7 +176,7 @@ func TestMigrateDatabase(t *testing.T) {
 				dBase, err := common.SqlOpen(dialect, u)
 				require.NoError(t, err)
 
-				man, err := schema.NewManager(ctx, dialect, dBase)
+				man, err := schema.NewManager(ctx, schema.Dialect(dialect), dBase)
 				require.NoError(t, err)
 				// This is an advisory lock on the DB which is released when the DB session ends.
 				err = man.ExclusiveLock(ctx)
@@ -220,9 +216,9 @@ func TestVerifyOplogIsEmpty(t *testing.T) {
 	dBase, err := common.SqlOpen(dialect, u)
 	require.NoError(t, err)
 
-	man, err := schema.NewManager(ctx, dialect, dBase)
+	man, err := schema.NewManager(ctx, schema.Dialect(dialect), dBase)
 	require.NoError(t, err)
-	require.NoError(t, man.RollForward(ctx))
+	require.NoError(t, man.ApplyMigrations(ctx))
 
 	cmd := InitCommand{Command: base.NewCommand(cli.NewMockUi())}
 	cmd.srv = base.NewServer(&base.Command{UI: cmd.UI})

internal/cmd/commands/server/server.go

@@ -425,24 +425,22 @@ func (c *Command) Run(args []string) int {
 			c.UI.Error(fmt.Errorf("Error checking schema state: %w", err).Error())
 			return base.CommandCliError
 		}
-		if !ckState.InitializationStarted {
+		if !ckState.Initialized {
 			c.UI.Error(base.WrapAtLength("The database has not been initialized. Please run 'boundary database init'."))
 			return base.CommandCliError
 		}
-		if ckState.Dirty {
-			c.UI.Error(base.WrapAtLength("Database is in a bad state. Please revert the database into the last known good state."))
-			return base.CommandCliError
-		}
-		if ckState.BinarySchemaVersion > ckState.DatabaseSchemaVersion {
+		if !ckState.MigrationsApplied() {
+			for _, e := range ckState.Editions {
+				if e.DatabaseSchemaState == schema.Ahead {
+					c.UI.Error(base.WrapAtLength(fmt.Sprintf("Newer schema version (%s %d) "+
+						"than this binary expects. Please use a newer version of the boundary "+
+						"binary.", e.Name, e.DatabaseSchemaVersion)))
+					return base.CommandCliError
+				}
+			}
 			c.UI.Error(base.WrapAtLength("Database schema must be updated to use this version. Run 'boundary database migrate' to update the database. NOTE: Boundary does not currently support live migration; ensure all controllers are shut down before running the migration command."))
 			return base.CommandCliError
 		}
-		if ckState.BinarySchemaVersion < ckState.DatabaseSchemaVersion {
-			c.UI.Error(base.WrapAtLength(fmt.Sprintf("Newer schema version (%d) "+
-				"than this binary expects. Please use a newer version of the boundary "+
-				"binary.", ckState.DatabaseSchemaVersion)))
-			return base.CommandCliError
-		}
 		if err := c.verifyKmsSetup(); err != nil {
 			c.UI.Error(base.WrapAtLength("Database is in a bad state. Please revert the database into the last known good state."))
 			return base.CommandCliError

internal/db/package_registration.go

@@ -0,0 +1,6 @@
+package db
+
+import (
+	// import for init side-effects to include migrations
+	_ "github.com/hashicorp/boundary/internal/db/schema/migrations/oss"
+)

internal/db/schema/doc.go

@@ -0,0 +1,22 @@
+// Package schema is used to apply sql migrations to modify the state of a
+// database instance.  It also provides functionality to report on the state of
+// a database instance and how it compares with the migration editions in the
+// running binary.
+//
+// This package requires that the migration editions are first registered,
+// prior to creating a schema.Manager. This is generally done in an init
+// function of another package, i.e.:
+//
+//     //go:embed postgres
+//     var postgres embed.FS
+//
+//     func init() {
+//         schema.RegisterEdition("oss", schema.Postgres, postgres, 0)
+//     }
+//
+// Then a manager can be created and used to apply the migrations:
+//
+//     m := schema.NewManager(ctx, schema.Postgres, db)
+//     m.ApplyMigrations()
+//
+package schema

internal/db/schema/edition.go

@@ -0,0 +1,68 @@
+package schema
+
+import (
+	"embed"
+	"fmt"
+	"sync"
+
+	"github.com/hashicorp/boundary/internal/db/schema/internal/edition"
+)
+
+// Dialect is same as edition.Dialect
+type Dialect = edition.Dialect
+
+// Supported dialects.
+const (
+	Postgres Dialect = "postgres"
+)
+
+var supportedDialects = map[Dialect]struct{}{
+	Postgres: struct{}{},
+}
+
+// type dialects map[Dialect]edition.Editions
+
+type dialects struct {
+	sync.Mutex
+
+	m map[Dialect]edition.Editions
+}
+
+var editions = dialects{
+	m: make(map[Dialect]edition.Editions),
+}
+
+// RegisterEdition registers an edition for use by the Manager. It will panic if:
+// - An unsupported dialect is provided.
+// - The same (dialect, name) is registered.
+// - The same (dialect, priority) is registered.
+func RegisterEdition(name string, dialect Dialect, fs embed.FS, priority int) {
+	editions.Lock()
+	defer editions.Unlock()
+
+	if _, ok := supportedDialects[dialect]; !ok {
+		panic(fmt.Sprintf("unsupported dialect: %s", dialect))
+	}
+	var e edition.Editions
+	var ok bool
+
+	e, ok = editions.m[dialect]
+	if !ok {
+		e = make(edition.Editions, 0)
+	}
+
+	for _, ee := range e {
+		if ee.Name == name {
+			panic(fmt.Sprintf("edition %s with dialect %s already registered", name, dialect))
+		}
+
+		if ee.Priority == priority {
+			panic(fmt.Sprintf("edition %s with dialect %s and priority %d has same priority as edition %s", name, dialect, priority, ee.Name))
+		}
+	}
+
+	e = append(e, edition.New(name, dialect, fs, priority))
+	e.Sort()
+
+	editions.m[dialect] = e
+}

internal/db/schema/edition_test.go

@@ -0,0 +1,161 @@
+package schema_test
+
+import (
+	"embed"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/db/schema"
+	"github.com/stretchr/testify/assert"
+)
+
+type testEdition struct {
+	name     string
+	dialect  schema.Dialect
+	fs       embed.FS
+	priority int
+}
+
+var (
+	//go:embed testdata/one
+	one embed.FS
+
+	//go:embed testdata/two
+	two embed.FS
+
+	//go:embed testdata/three
+	three embed.FS
+)
+
+func TestRegisterEditionPanics(t *testing.T) {
+	tests := []struct {
+		name     string
+		editions []testEdition
+	}{
+		{
+			"unsupportedDialect",
+			[]testEdition{
+				{
+					"one",
+					schema.Dialect("mongodb"),
+					one,
+					0,
+				},
+			},
+		},
+		{
+			"duplicateName",
+			[]testEdition{
+				{
+					"one",
+					schema.Postgres,
+					one,
+					0,
+				},
+				{
+					"one",
+					schema.Postgres,
+					one,
+					1,
+				},
+			},
+		},
+		{
+			"duplicatePriority",
+			[]testEdition{
+				{
+					"one",
+					schema.Postgres,
+					one,
+					0,
+				},
+				{
+					"two",
+					schema.Postgres,
+					two,
+					0,
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Panics(t, func() {
+				for _, e := range tt.editions {
+					schema.RegisterEdition(e.name, e.dialect, e.fs, e.priority)
+				}
+			}, tt.name)
+		})
+	}
+}
+
+func TestRegisterEdition(t *testing.T) {
+	tests := []struct {
+		name     string
+		editions []testEdition
+	}{
+		{
+			"singleEdition",
+			[]testEdition{
+				{
+					"one",
+					schema.Postgres,
+					one,
+					0,
+				},
+			},
+		},
+		{
+			"twoEditions",
+			[]testEdition{
+				{
+					"one",
+					schema.Postgres,
+					one,
+					0,
+				},
+				{
+					"two",
+					schema.Postgres,
+					one,
+					1,
+				},
+			},
+		},
+		{
+			"threeEditions",
+			[]testEdition{
+				{
+					"one",
+					schema.Postgres,
+					one,
+					0,
+				},
+				{
+					"two",
+					schema.Postgres,
+					one,
+					1,
+				},
+				{
+					"three",
+					schema.Postgres,
+					one,
+					3,
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.NotPanics(t, func() {
+				schema.TestClearEditions(t)
+				for _, e := range tt.editions {
+					schema.RegisterEdition(e.name, e.dialect, e.fs, e.priority)
+				}
+			}, tt.name)
+		})
+	}
+	schema.TestClearEditions(t)
+}

internal/db/schema/export_test.go

@@ -0,0 +1,18 @@
+package schema
+
+import (
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/db/schema/internal/edition"
+)
+
+const NilVersion = nilVersion
+
+// TestClearEditions is a test helper to reset the editions map.
+func TestClearEditions(t *testing.T) {
+	t.Helper()
+	editions.Lock()
+	defer editions.Unlock()
+
+	editions.m = make(map[edition.Dialect]edition.Editions)
+}

internal/db/schema/internal/edition/edition.go

@@ -0,0 +1,136 @@
+// Package edition provides internal structs for the schema package for
+// defining and organizing database migration editions.
+package edition
+
+import (
+	"embed"
+	"fmt"
+	"io/fs"
+	"path/filepath"
+	"sort"
+	"strconv"
+	"strings"
+)
+
+// Dialect is a specific SQL language variant. This generally is the same as
+// a specific SQL server implementation.
+type Dialect string
+
+func (d Dialect) String() string {
+	return string(d)
+}
+
+// Edition is a collection of sql statements along with a version number.
+// It is used to apply changes to the database instance.
+type Edition struct {
+	// The identifier of an edition. The Name and Dialect pair should be unique.
+	Name string
+	// The specific SQL language variant that the Migrtions are for. The Name
+	// and Dialect pair should be unique.
+	Dialect Dialect
+
+	// The latest version of the schema for this edition that this binary supports.
+	LatestVersion int
+
+	// The set of migrations that should be applied to a database to reach the latest version.
+	// This is a map of schema versions to sql.
+	Migrations map[int][]byte
+
+	// Priority is used to determine the order that multiple Editions should be applied.
+	Priority int
+}
+
+// Editions is a collection of Edition
+type Editions []Edition
+
+// Sort orders the Editions by priority.
+func (e Editions) Sort() {
+	sort.SliceStable(e, func(i, j int) bool {
+		return e[i].Priority < e[j].Priority
+	})
+}
+
+// New creates an Edition with the provided parameters. The embed.FS m will be
+// walked to extract the sql statements. The priority is used to determine
+// when this Edition's migrations are applied relative to other Editions. A
+// lower number indicates a higher priority. New will panic if the structure
+// of the embed.FS is not correct. The files must be structured as follows:
+//
+//   <majorVersion>/
+//       <minorVersion>_<description>.up.sql
+//
+// Where majorVersion and minorVersion are integers. There can be any number of
+// leading directories prior to the major versions. For example a directory
+// structure like the following is correct:
+//
+//   migrations/oss/postgres/
+//    0/
+//      01_initial.up.sql
+//    1/
+//      01_add_columns.up.sql
+//      02_rename_table.up.sql
+//    2/
+//      01_add_new_table.up.sql
+//      02_refactor_views.up.sql
+func New(name string, dialect Dialect, m embed.FS, priority int) Edition {
+	var largestSchemaVersion int
+	migrations := make(map[int][]byte)
+
+	fs.WalkDir(m, ".", func(path string, d fs.DirEntry, err error) error {
+		if err != nil {
+			panic(fmt.Sprintf("unable to process migration files: %s", err))
+		}
+
+		if d.IsDir() {
+			return nil
+		}
+
+		if !strings.HasSuffix(path, ".up.sql") {
+			return nil
+		}
+
+		dir, file := filepath.Split(path)
+		verMajorDir := filepath.Base(dir)
+
+		verMajor, err := strconv.Atoi(verMajorDir)
+		if err != nil {
+			panic(fmt.Sprintf("migration file does not have valid major version directory: %s", path))
+		}
+
+		verMinor, err := strconv.Atoi(strings.SplitN(file, "_", 2)[0])
+		if err != nil {
+			panic(fmt.Sprintf("migration file does not have valid minor version prefix: %s", path))
+		}
+
+		fullV := (verMajor * 1000) + verMinor
+		if fullV > largestSchemaVersion {
+			largestSchemaVersion = fullV
+		}
+
+		cbts, err := m.ReadFile(path)
+		if err != nil {
+			panic(fmt.Sprintf("unable to read migration file: %s", path))
+		}
+
+		contents := strings.TrimSpace(string(cbts))
+		if strings.ToLower(contents[:len("begin;")]) == "begin;" {
+			contents = contents[len("begin;"):]
+		}
+		if strings.ToLower(contents[len(contents)-len("commit;"):]) == "commit;" {
+			contents = contents[:len(contents)-len("commit;")]
+		}
+		contents = strings.TrimSpace(contents)
+
+		migrations[fullV] = []byte(contents)
+
+		return nil
+	})
+
+	return Edition{
+		Name:          name,
+		Dialect:       dialect,
+		LatestVersion: largestSchemaVersion,
+		Migrations:    migrations,
+		Priority:      priority,
+	}
+}

internal/db/schema/internal/edition/edition_test.go

@@ -0,0 +1,264 @@
+package edition_test
+
+import (
+	"embed"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/db/schema/internal/edition"
+	"github.com/stretchr/testify/assert"
+)
+
+// valid embed.FS
+var (
+	//go:embed testdata/one
+	one embed.FS
+	//go:embed testdata/two
+	two embed.FS
+	//go:embed testdata/three
+	three embed.FS
+)
+
+func TestNew(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name                   string
+		priority               int
+		fs                     embed.FS
+		expectedVersion        int
+		expectedMigrationCount int
+	}{
+		{
+			"one",
+			0,
+			one,
+			1,
+			1,
+		},
+		{
+			"two",
+			1,
+			two,
+			2,
+			2,
+		},
+		{
+			"three",
+			3,
+			three,
+			1001,
+			2,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			e := edition.New(tt.name, edition.Dialect("postgres"), tt.fs, tt.priority)
+			assert.Equal(t, e.Name, tt.name, "Name")
+			assert.Equal(t, e.Dialect, edition.Dialect("postgres"), "Dialect")
+			assert.Equal(t, e.LatestVersion, tt.expectedVersion, "Version")
+			assert.Equal(t, e.Priority, tt.priority, "Priority")
+			assert.Equal(t, len(e.Migrations), tt.expectedMigrationCount, "Number of migrations")
+		})
+	}
+}
+
+// invalid embed.FS
+var (
+	//go:embed testdata/invalid/major-version-not-int
+	majorVersionNotInt embed.FS
+	//go:embed testdata/invalid/minor-version-invalid-separator
+	minorVersionInvalidSeparator embed.FS
+	//go:embed testdata/invalid/minor-version-not-int
+	minorVersionNotInt embed.FS
+	//go:embed testdata/invalid/no-minor-version
+	noMinorVersion embed.FS
+	//go:embed testdata/invalid/no-major-version
+	noMajorVersion embed.FS
+)
+
+func TestNewPanics(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		fs   embed.FS
+	}{
+		{
+			"majorVersionNotInt",
+			majorVersionNotInt,
+		},
+		{
+			"minorVersionInvalidSeparator",
+			minorVersionInvalidSeparator,
+		},
+		{
+			"minorVersionNotInt",
+			minorVersionNotInt,
+		},
+		{
+			"noMinorVersion",
+			noMinorVersion,
+		},
+		{
+			"noMajorVersion",
+			noMajorVersion,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Panics(t, func() {
+				edition.New(tt.name, edition.Dialect("postgres"), tt.fs, 0)
+			}, tt.name)
+		})
+	}
+}
+
+func TestSort(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name     string
+		editions edition.Editions
+		sorted   edition.Editions
+	}{
+		{
+			"unsortedConsecutive",
+			edition.Editions{
+				{
+					Name:     "two",
+					Priority: 2,
+				},
+				{
+					Name:     "zero",
+					Priority: 0,
+				},
+				{
+					Name:     "one",
+					Priority: 1,
+				},
+			},
+			edition.Editions{
+				{
+					Name:     "zero",
+					Priority: 0,
+				},
+				{
+					Name:     "one",
+					Priority: 1,
+				},
+				{
+					Name:     "two",
+					Priority: 2,
+				},
+			},
+		},
+		{
+			"unsorted",
+			edition.Editions{
+				{
+					Name:     "three",
+					Priority: 3,
+				},
+				{
+					Name:     "zero",
+					Priority: 0,
+				},
+				{
+					Name:     "one",
+					Priority: 1,
+				},
+			},
+			edition.Editions{
+				{
+					Name:     "zero",
+					Priority: 0,
+				},
+				{
+					Name:     "one",
+					Priority: 1,
+				},
+				{
+					Name:     "three",
+					Priority: 3,
+				},
+			},
+		},
+		{
+			"equalPriority",
+			edition.Editions{
+				{
+					Name:     "three-also",
+					Priority: 3,
+				},
+				{
+					Name:     "three",
+					Priority: 3,
+				},
+				{
+					Name:     "zero",
+					Priority: 0,
+				},
+				{
+					Name:     "one",
+					Priority: 1,
+				},
+			},
+			edition.Editions{
+				{
+					Name:     "zero",
+					Priority: 0,
+				},
+				{
+					Name:     "one",
+					Priority: 1,
+				},
+				{
+					Name:     "three-also",
+					Priority: 3,
+				},
+				{
+					Name:     "three",
+					Priority: 3,
+				},
+			},
+		},
+		{
+			"alreadySorted",
+			edition.Editions{
+				{
+					Name:     "zero",
+					Priority: 0,
+				},
+				{
+					Name:     "one",
+					Priority: 1,
+				},
+				{
+					Name:     "two",
+					Priority: 2,
+				},
+			},
+			edition.Editions{
+				{
+					Name:     "zero",
+					Priority: 0,
+				},
+				{
+					Name:     "one",
+					Priority: 1,
+				},
+				{
+					Name:     "two",
+					Priority: 2,
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.editions.Sort()
+			assert.Equal(t, tt.editions, tt.sorted)
+		})
+	}
+}

internal/db/schema/internal/edition/testdata/invalid/major-version-not-int/major/01_initial.up.sql

@@ -0,0 +1,5 @@
+begin;
+  create table one (
+    id bigint primary key
+  );
+commit;

internal/db/schema/internal/edition/testdata/invalid/minor-version-invalid-separator/0/01-initial.up.sql

@@ -0,0 +1,5 @@
+begin;
+  create table one (
+    id bigint primary key
+  );
+commit;

internal/db/schema/internal/edition/testdata/invalid/minor-version-not-int/0/minor_initial.up.sql

@@ -0,0 +1,5 @@
+begin;
+  create table one (
+    id bigint primary key
+  );
+commit;

internal/db/schema/internal/edition/testdata/invalid/no-major-version/01_initial.up.sql

@@ -0,0 +1,5 @@
+begin;
+  create table one (
+    id bigint primary key
+  );
+commit;

internal/db/schema/internal/edition/testdata/invalid/no-minor-version/0/initial.up.sql

@@ -0,0 +1,5 @@
+begin;
+  create table one (
+    id bigint primary key
+  );
+commit;

internal/db/schema/internal/edition/testdata/one/0/01_initial.up.sql

@@ -0,0 +1,5 @@
+begin;
+  create table one (
+    id bigint primary key
+  );
+commit;

internal/db/schema/internal/edition/testdata/three/0/01_initial.up.sql

@@ -0,0 +1,5 @@
+begin;
+  create table three (
+    id bigint primary key
+  );
+commit;

internal/db/schema/internal/edition/testdata/three/1/01_add_table.up.sql

@@ -0,0 +1,3 @@
+begin;
+  alter table tree add column name text;
+commit;

internal/db/schema/internal/edition/testdata/two/0/01_add_table.up.sql

@@ -0,0 +1,5 @@
+begin;
+  create table two (
+    id bigint primary key
+  );
+commit;

internal/db/schema/internal/edition/testdata/two/0/02_initial.up.sql

@@ -0,0 +1,3 @@
+begin;
+  alter table two add column name text;
+commit;

internal/db/schema/internal/log/doc.go

@@ -0,0 +1,3 @@
+// Package log provides internal structs and options for the schema package for
+// tracking logs generated when applying migrations.
+package log

internal/db/schema/internal/log/entry.go

@@ -0,0 +1,12 @@
+package log
+
+import "time"
+
+// Entry represents a log entry generated during migrations.
+type Entry struct {
+	Id               int
+	MigrationEdition string
+	MigrationVersion int
+	CreateTime       time.Time
+	Entry            string
+}

internal/db/schema/internal/log/options.go

@@ -0,0 +1,29 @@
+package log
+
+// GetOpts - iterate the inbound Options and return a struct.
+func GetOpts(opt ...Option) Options {
+	opts := getDefaultOptions()
+	for _, o := range opt {
+		o(&opts)
+	}
+	return opts
+}
+
+// Option - how Options are passed as arguments.
+type Option func(*Options)
+
+// Options = how options are represented
+type Options struct {
+	WithDeleteLog bool
+}
+
+func getDefaultOptions() Options {
+	return Options{}
+}
+
+// WithDeleteLog provides an option to specify the deletion of log entries.
+func WithDeleteLog(del bool) Option {
+	return func(o *Options) {
+		o.WithDeleteLog = del
+	}
+}

internal/db/schema/internal/postgres/doc.go

@@ -0,0 +1,3 @@
+// Package postgres provides an implementation of the schema.driver interface
+// for a PostgreSQL database.
+package postgres

internal/db/schema/internal/postgres/ensure_version_test.go

@@ -0,0 +1,111 @@
+package postgres_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEnsureVersion(t *testing.T) {
+	ctx := context.Background()
+	p, _, _ := setup(ctx, t)
+
+	err := p.StartRun(ctx)
+	require.NoError(t, err)
+
+	err = p.EnsureVersionTable(ctx)
+	require.NoError(t, err)
+
+	err = p.CommitRun(ctx)
+	require.NoError(t, err)
+}
+
+func TestEnsureVersion_NoTxn(t *testing.T) {
+	ctx := context.Background()
+	p, _, _ := setup(ctx, t)
+	err := p.EnsureVersionTable(ctx)
+	require.Error(t, err)
+}
+
+func TestEnsureVersion_UpdateForEditionSupport(t *testing.T) {
+	ctx := context.Background()
+	p, db, _ := setup(ctx, t)
+
+	_, err := db.ExecContext(ctx, `
+	create table boundary_schema_version (
+		version bigint primary key,
+		dirty boolean not null
+	);`)
+	require.NoError(t, err)
+	_, err = db.ExecContext(ctx, `
+	insert into boundary_schema_version
+	(version, dirty)
+	values
+	(1001, false);`)
+	require.NoError(t, err)
+
+	err = p.StartRun(ctx)
+	require.NoError(t, err)
+
+	err = p.EnsureVersionTable(ctx)
+	require.NoError(t, err)
+
+	err = p.CommitRun(ctx)
+	require.NoError(t, err)
+
+	var edition string
+	var version int
+
+	err = db.QueryRowContext(
+		ctx,
+		`select edition, version from boundary_schema_version`,
+	).Scan(&edition, &version)
+
+	require.NoError(t, err)
+
+	assert.Equal(t, edition, "oss")
+	assert.Equal(t, version, 1001)
+}
+
+func TestEnsureVersion_RenameOldSchemaTable(t *testing.T) {
+	ctx := context.Background()
+	p, db, _ := setup(ctx, t)
+
+	_, err := db.ExecContext(ctx, `
+	create table schema_migrations (
+		version bigint primary key,
+		dirty boolean not null
+	);`)
+	require.NoError(t, err)
+	require.NoError(t, err)
+	_, err = db.ExecContext(ctx, `
+	insert into schema_migrations
+	(version, dirty)
+	values
+	(1001, false);`)
+	require.NoError(t, err)
+
+	err = p.StartRun(ctx)
+	require.NoError(t, err)
+
+	err = p.EnsureVersionTable(ctx)
+	require.NoError(t, err)
+
+	err = p.CommitRun(ctx)
+	require.NoError(t, err)
+
+	var edition string
+	var version int
+
+	err = db.QueryRowContext(
+		ctx,
+		`select edition, version from boundary_schema_version`,
+	).Scan(&edition, &version)
+
+	require.NoError(t, err)
+
+	assert.Equal(t, edition, "oss")
+	assert.Equal(t, version, 1001)
+}

internal/db/schema/internal/postgres/error_processing.go

@@ -0,0 +1,66 @@
+// The MIT License (MIT)
+//
+// Original Work
+// Copyright (c) 2016 Matthias Kadenbach
+// https://github.com/mattes/migrate
+//
+// Modified Work
+// Copyright (c) 2018 Dale Hui
+// https://github.com/golang-migrate/migrate
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package postgres
+
+import "strings"
+
+const newLine = '\n'
+
+func runesCount(input []rune, target rune) int {
+	var count int
+	for _, r := range input {
+		if r == target {
+			count++
+		}
+	}
+	return count
+}
+
+func runesLastIndex(input []rune, target rune) int {
+	for i := len(input) - 1; i >= 0; i-- {
+		if input[i] == target {
+			return i
+		}
+	}
+	return -1
+}
+
+func computeLineFromPos(s string, pos int) (line uint, col uint, ok bool) {
+	// replace crlf with lf
+	s = strings.Replace(s, "\r\n", "\n", -1)
+	// pg docs: pos uses index 1 for the first character, and positions are measured in characters not bytes
+	runes := []rune(s)
+	if pos > len(runes) {
+		return 0, 0, false
+	}
+	sel := runes[:pos]
+	line = uint(runesCount(sel, newLine) + 1)
+	col = uint(pos - 1 - runesLastIndex(sel, newLine))
+	return line, col, true
+}

internal/db/schema/internal/postgres/error_processing_test.go

@@ -0,0 +1,115 @@
+// The MIT License (MIT)
+//
+// Original Work
+// Copyright (c) 2016 Matthias Kadenbach
+// https://github.com/mattes/migrate
+//
+// Modified Work
+// Copyright (c) 2018 Dale Hui
+// https://github.com/golang-migrate/migrate
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package postgres
+
+import (
+	"strconv"
+	"strings"
+	"testing"
+)
+
+func Test_computeLineFromPos(t *testing.T) {
+	testcases := []struct {
+		pos      int
+		wantLine uint
+		wantCol  uint
+		input    string
+		wantOk   bool
+	}{
+		{
+			15, 2, 6, "SELECT *\nFROM foo", true, // foo table does not exists
+		},
+		{
+			16, 3, 6, "SELECT *\n\nFROM foo", true, // foo table does not exists, empty line
+		},
+		{
+			25, 3, 7, "SELECT *\nFROM foo\nWHERE x", true, // x column error
+		},
+		{
+			27, 5, 7, "SELECT *\n\nFROM foo\n\nWHERE x", true, // x column error, empty lines
+		},
+		{
+			10, 2, 1, "SELECT *\nFROMM foo", true, // FROMM typo
+		},
+		{
+			11, 3, 1, "SELECT *\n\nFROMM foo", true, // FROMM typo, empty line
+		},
+		{
+			17, 2, 8, "SELECT *\nFROM foo", true, // last character
+		},
+		{
+			18, 0, 0, "SELECT *\nFROM foo", false, // invalid position
+		},
+	}
+	for i, tc := range testcases {
+		t.Run("tc"+strconv.Itoa(i), func(t *testing.T) {
+			run := func(crlf bool, nonASCII bool) {
+				var name string
+				if crlf {
+					name = "crlf"
+				} else {
+					name = "lf"
+				}
+				if nonASCII {
+					name += "-nonascii"
+				} else {
+					name += "-ascii"
+				}
+				t.Run(name, func(t *testing.T) {
+					input := tc.input
+					if crlf {
+						input = strings.Replace(input, "\n", "\r\n", -1)
+					}
+					if nonASCII {
+						input = strings.Replace(input, "FROM", "FRÖM", -1)
+					}
+					gotLine, gotCol, gotOK := computeLineFromPos(input, tc.pos)
+
+					if tc.wantOk {
+						t.Logf("pos %d, want %d:%d, %#v", tc.pos, tc.wantLine, tc.wantCol, input)
+					}
+
+					if gotOK != tc.wantOk {
+						t.Fatalf("expected ok %v but got %v", tc.wantOk, gotOK)
+					}
+					if gotLine != tc.wantLine {
+						t.Fatalf("expected line %d but got %d", tc.wantLine, gotLine)
+					}
+					if gotCol != tc.wantCol {
+						t.Fatalf("expected col %d but got %d", tc.wantCol, gotCol)
+					}
+				})
+			}
+			run(false, false)
+			run(true, false)
+			run(false, true)
+			run(true, true)
+		})
+	}
+}

internal/db/schema/internal/postgres/lock_test.go

@@ -0,0 +1,152 @@
+package postgres_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/boundary/internal/db/common"
+	"github.com/hashicorp/boundary/internal/db/schema/internal/postgres"
+	"github.com/hashicorp/boundary/testing/dbtest"
+	"github.com/stretchr/testify/require"
+)
+
+// Tests that TrySharedLock:
+// - can get a lock as the only connection
+// - a second connection can also get the shared lock
+// - a second connection cannot get an exclusive lock
+func TestTrySharedLock(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	p, _, u := setup(ctx, t)
+
+	err := p.TrySharedLock(ctx)
+	require.NoError(t, err)
+
+	d2, err := common.SqlOpen(dbtest.Postgres, u)
+	p2, err := postgres.New(ctx, d2)
+	require.NoError(t, err)
+
+	err = p2.TrySharedLock(ctx)
+	require.NoError(t, err)
+
+	err = p2.TryLock(ctx)
+	require.Error(t, err)
+}
+
+// Test that TryLock gets an exclusive lock and:
+// - a separate connection cannot get the lock via TryLock
+// - a separate connection cannot get the lock via TrySharedLock
+// - a separate connection cannot get the lock via Lock
+func TestTryLock(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	p, _, u := setup(ctx, t)
+
+	err := p.TryLock(ctx)
+	require.NoError(t, err)
+
+	d2, err := common.SqlOpen(dbtest.Postgres, u)
+	p2, err := postgres.New(ctx, d2)
+	require.NoError(t, err)
+
+	err = p2.TrySharedLock(ctx)
+	require.Error(t, err)
+
+	err = p2.TryLock(ctx)
+	require.Error(t, err)
+
+	ctx2, _ := context.WithTimeout(ctx, time.Millisecond)
+	err = p2.Lock(ctx2)
+	require.Error(t, err)
+}
+
+// Test that Lock gets an exclusive lock and:
+// - a separate connection cannot get the lock via TryLock
+// - a separate connection cannot get the lock via TrySharedLock
+// - a separate connection cannot get the lock via Lock
+func TestLock(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	p, _, u := setup(ctx, t)
+
+	err := p.Lock(ctx)
+	require.NoError(t, err)
+
+	d2, err := common.SqlOpen(dbtest.Postgres, u)
+	p2, err := postgres.New(ctx, d2)
+	require.NoError(t, err)
+
+	err = p2.TrySharedLock(ctx)
+	require.Error(t, err)
+
+	err = p2.TryLock(ctx)
+	require.Error(t, err)
+
+	ctx2, _ := context.WithTimeout(ctx, time.Millisecond)
+	err = p2.Lock(ctx2)
+	require.Error(t, err)
+}
+
+// Test the Unlock:
+// - can unlock a lock granted via TryLock
+// - can unlock a lock granted via Lock
+func TestUnlock(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	p, _, u := setup(ctx, t)
+
+	err := p.Lock(ctx)
+	require.NoError(t, err)
+
+	d2, err := common.SqlOpen(dbtest.Postgres, u)
+	p2, err := postgres.New(ctx, d2)
+	require.NoError(t, err)
+
+	// cannot get lock since p has it.
+	err = p2.TryLock(ctx)
+	require.Error(t, err)
+
+	// clear lock on p
+	err = p.Unlock(ctx)
+	require.NoError(t, err)
+
+	// p2 can now get lock
+	err = p2.TryLock(ctx)
+	require.NoError(t, err)
+
+	// p can not get the lock now
+	err = p.TryLock(ctx)
+	require.Error(t, err)
+
+	// clear lock on p2
+	err = p2.Unlock(ctx)
+
+	// p can now get lock
+	err = p.TryLock(ctx)
+	require.NoError(t, err)
+}
+
+// Test that UnlockShared
+// - can unlock a lock granted by TrySharedLock
+func TestUnlockShared(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	p, _, u := setup(ctx, t)
+
+	err := p.TrySharedLock(ctx)
+	require.NoError(t, err)
+
+	d2, err := common.SqlOpen(dbtest.Postgres, u)
+	p2, err := postgres.New(ctx, d2)
+	require.NoError(t, err)
+
+	err = p2.TryLock(ctx)
+	require.Error(t, err)
+
+	err = p.UnlockShared(ctx)
+	require.NoError(t, err)
+
+	err = p2.TryLock(ctx)
+	require.NoError(t, err)
+}

internal/db/schema/internal/postgres/migration_log_test.go

@@ -0,0 +1,203 @@
+package postgres_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/db/schema/internal/log"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEnsureMigrationLog(t *testing.T) {
+	ctx := context.Background()
+	p, _, _ := setup(ctx, t)
+
+	err := p.StartRun(ctx)
+	require.NoError(t, err)
+
+	err = p.EnsureVersionTable(ctx)
+	require.NoError(t, err)
+
+	err = p.EnsureMigrationLogTable(ctx)
+	require.NoError(t, err)
+
+	err = p.CommitRun(ctx)
+	require.NoError(t, err)
+}
+
+func TestEnsureMigrationLog_NoTxn(t *testing.T) {
+	ctx := context.Background()
+	p, _, _ := setup(ctx, t)
+	err := p.EnsureMigrationLogTable(ctx)
+	require.Error(t, err)
+}
+
+// Test that a database that was created with a log_migration table
+// prior to edition support, can be updated to support editions,
+// and any existing log entries can be updated.
+//
+// The scenario involves a case where:
+// - The database was initialized with the original boundary_schema_version table
+// - The original migrations where run, creating the log_migration table and some entries in the table.
+func TestEnsureMigrationLog_UpdateForEditionSupport(t *testing.T) {
+	ctx := context.Background()
+	p, db, _ := setup(ctx, t)
+
+	statements := []string{
+		// original boundary_schema_version table
+		`create table boundary_schema_version (
+			 version bigint primary key,
+			 dirty boolean not null);`,
+		// populate the table with a version
+		`insert into boundary_schema_version
+			 (version, dirty)
+		 values
+			 (1001, false);`,
+		// need to create some things from other migrations that the original tables used
+		`create domain wt_timestamp as
+			 timestamp with time zone
+			 default current_timestamp;
+
+			create or replace function
+				default_create_time()
+				returns trigger
+			as $$
+			begin
+				if new.create_time is distinct from now() then
+					raise warning 'create_time cannot be set to %', new.create_time;
+					new.create_time = now();
+				end if;
+				return new;
+			end;
+			$$ language plpgsql;
+
+			create or replace function
+				immutable_columns()
+				returns trigger
+			as $$
+			declare
+				col_name text;
+				new_value text;
+				old_value text;
+			begin
+				foreach col_name in array tg_argv loop
+					execute format('SELECT $1.%I', col_name) into new_value using new;
+					execute format('SELECT $1.%I', col_name) into old_value using old;
+					if new_value is distinct from old_value then
+						raise exception 'immutable column: %.%', tg_table_name, col_name using
+							errcode = '23601',
+							schema = tg_table_schema,
+							table = tg_table_name,
+							column = col_name;
+					end if;
+				end loop;
+				return new;
+			end;
+			$$ language plpgsql;
+		`,
+		// create original log_migration table, triggers
+		`create table log_migration(
+			id bigint generated always as identity primary key,
+			migration_version bigint not null, -- cannot declare FK since the table is truncated during runtime
+			create_time wt_timestamp,
+			entry text not null
+		 );
+		 comment on table log_migration is
+		 'log_migration entries are logging output from databaes migrations';
+
+		 create trigger
+			default_create_time_column
+		 before
+		 insert on log_migration
+			for each row execute procedure default_create_time();
+
+		 create trigger
+			immutable_columns
+		 before
+		 update on log_migration
+			for each row execute procedure immutable_columns('id', 'migration_version', 'create_time', 'entry');
+
+		 create or replace function
+			log_migration_version()
+			returns trigger
+		 as $$
+			declare current_version bigint;
+			begin
+				select max(version) from boundary_schema_version into current_version;
+				new.migration_version = current_version;
+				return new;
+			end;
+		 $$ language plpgsql;
+		 comment on function log_migration_version() is
+		 'log_migration_version will set the log_migration entries to the current migration version';
+
+		 create trigger
+			migration_version_column
+		 before
+		 insert on log_migration
+			for each row execute procedure log_migration_version();
+		`,
+		// insert some logs
+		`insert into log_migration (entry)
+		values
+		 ('log entry');`,
+	}
+
+	for _, s := range statements {
+		_, err := db.ExecContext(ctx, s)
+		require.NoError(t, err)
+	}
+
+	err := p.StartRun(ctx)
+	require.NoError(t, err)
+
+	err = p.EnsureVersionTable(ctx)
+	require.NoError(t, err)
+
+	err = p.EnsureMigrationLogTable(ctx)
+	require.NoError(t, err)
+
+	err = p.CommitRun(ctx)
+	require.NoError(t, err)
+
+	logs, err := p.GetMigrationLog(ctx, log.WithDeleteLog(true))
+	require.NoError(t, err)
+	assert.Equal(t, 1, len(logs))
+	assert.NotEqual(t, logs[0].Id, 0)
+	assert.False(t, logs[0].CreateTime.IsZero())
+	assert.Equal(t, "oss", logs[0].MigrationEdition)
+	assert.Equal(t, 1001, logs[0].MigrationVersion)
+	assert.Equal(t, "log entry", logs[0].Entry)
+
+	// now ensure the updated table and trigger still works
+	statements = []string{
+		// update the version, to make it easy to filter on log_migrations
+		`update boundary_schema_version set version = 1002;`,
+		// insert a new log entry
+		`insert into log_migration (entry, edition)
+		values
+		 ('new log entry', 'oss');`,
+	}
+	for _, s := range statements {
+		_, err := db.ExecContext(ctx, s)
+		require.NoError(t, err)
+	}
+
+	logs, err = p.GetMigrationLog(ctx, log.WithDeleteLog(true))
+	require.NoError(t, err)
+	assert.Equal(t, 1, len(logs))
+	assert.NotEqual(t, logs[0].Id, 0)
+	assert.False(t, logs[0].CreateTime.IsZero())
+	assert.Equal(t, "oss", logs[0].MigrationEdition)
+	assert.Equal(t, 1002, logs[0].MigrationVersion)
+	assert.Equal(t, "new log entry", logs[0].Entry)
+
+	// ensure it is idempotent
+	err = p.StartRun(ctx)
+	require.NoError(t, err)
+	err = p.EnsureMigrationLogTable(ctx)
+	require.NoError(t, err)
+	err = p.CommitRun(ctx)
+	require.NoError(t, err)
+}

internal/db/schema/internal/postgres/postgres.go

@@ -0,0 +1,457 @@
+// The MIT License (MIT)
+//
+// Original Work
+// Copyright (c) 2016 Matthias Kadenbach
+// https://github.com/mattes/migrate
+//
+// Modified Work
+// Copyright (c) 2018 Dale Hui
+// https://github.com/golang-migrate/migrate
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package postgres
+
+import (
+	"context"
+	"database/sql"
+	stderrors "errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+
+	"github.com/hashicorp/boundary/internal/db/schema/internal/log"
+	"github.com/hashicorp/boundary/internal/db/schema/migrations"
+	"github.com/hashicorp/boundary/internal/errors"
+	"github.com/hashicorp/go-multierror"
+	"github.com/jackc/pgconn"
+)
+
+// schemaAccessLockId is a Lock key used to ensure a single boundary binary is operating
+// on a postgres server at a time.  The value has no meaning and was picked randomly.
+const schemaAccessLockId int64 = 3865661975
+
+// nilVersion is used to identify when a migration version has not be set.
+const nilVersion = -1
+
+// Postgres is a driver usable by a boundary schema.Manager.
+// This struct is not thread safe.
+type Postgres struct {
+	// Locking and unlocking need to use the same connection
+	conn *sql.Conn
+
+	tx *sql.Tx
+}
+
+// New creates a Postgres with the provided sql.DB verified as connectable
+func New(ctx context.Context, db *sql.DB) (*Postgres, error) {
+	const op = "postgres.New"
+	if err := db.PingContext(ctx); err != nil {
+		return nil, errors.Wrap(ctx, err, op)
+	}
+	conn, err := db.Conn(ctx)
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op)
+	}
+
+	px := &Postgres{
+		conn: conn,
+	}
+	return px, nil
+}
+
+// TrySharedLock attempts to capture a shared lock. If it is not successful it returns an error.
+// https://www.postgresql.org/docs/11/static/explicit-locking.html#ADVISORY-LOCKS
+func (p *Postgres) TrySharedLock(ctx context.Context) error {
+	const op = "postgres.(Postgres).TrySharedLock"
+
+	r := p.conn.QueryRowContext(ctx, trySharedLock, schemaAccessLockId)
+	if r.Err() != nil {
+		return errors.Wrap(ctx, r.Err(), op)
+	}
+	var gotLock bool
+	if err := r.Scan(&gotLock); err != nil {
+		return errors.Wrap(ctx, err, op)
+	}
+	if !gotLock {
+		return errors.New(ctx, errors.MigrationLock, op, "Lock failed")
+	}
+	return nil
+}
+
+// TryLock attempts to capture an exclusive lock. If it is not successful it returns an error.
+// https://www.postgresql.org/docs/11/static/explicit-locking.html#ADVISORY-LOCKS
+func (p *Postgres) TryLock(ctx context.Context) error {
+	const op = "postgres.(Postgres).TryLock"
+
+	r := p.conn.QueryRowContext(ctx, tryLock, schemaAccessLockId)
+	if r.Err() != nil {
+		return errors.Wrap(ctx, r.Err(), op)
+	}
+	var gotLock bool
+	if err := r.Scan(&gotLock); err != nil {
+		return errors.Wrap(ctx, err, op)
+	}
+	if !gotLock {
+		return errors.New(ctx, errors.MigrationLock, op, "Lock failed")
+	}
+	return nil
+}
+
+// Lock calls pg_advisory_lock with the provided context and returns an error
+// if we were unable to get the lock before the context cancels.
+func (p *Postgres) Lock(ctx context.Context) error {
+	const op = "postgres.(Postgres).Lock"
+
+	if _, err := p.conn.ExecContext(ctx, lock, schemaAccessLockId); err != nil {
+		return errors.Wrap(ctx, err, op)
+	}
+	return nil
+}
+
+// Unlock calls pg_advisory_unlock and returns an error if we were unable to
+// release the lock before the context cancels.
+func (p *Postgres) Unlock(ctx context.Context) error {
+	const op = "postgres.(Postgres).Unlock"
+
+	if _, err := p.conn.ExecContext(ctx, unlock, schemaAccessLockId); err != nil {
+		return errors.Wrap(ctx, err, op)
+	}
+	return nil
+}
+
+// UnlockShared calls pg_advisory_unlock_shared and returns an error if we were unable to
+// release the lock before the context cancels.
+func (p *Postgres) UnlockShared(ctx context.Context) error {
+	const op = "postgres.(Postgres).UnlockShared"
+
+	if _, err := p.conn.ExecContext(ctx, unlockSharedLock, schemaAccessLockId); err != nil {
+		return errors.Wrap(ctx, err, op)
+	}
+	return nil
+}
+
+// StartRun begins a transaction internal to the driver.
+func (p *Postgres) StartRun(ctx context.Context) error {
+	tx, err := p.conn.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	p.tx = tx
+	return nil
+}
+
+// CommitRun commits a transaction, if there is an error it should rollback the transaction.
+func (p *Postgres) CommitRun(ctx context.Context) error {
+	const op = "postgres.(Postgres).CommitRun"
+	defer func() {
+		p.tx = nil
+	}()
+	if p.tx == nil {
+		return errors.New(ctx, errors.MigrationIntegrity, op, "no pending transaction")
+	}
+	if err := p.tx.Commit(); err != nil {
+		if errRollback := p.tx.Rollback(); errRollback != nil && errRollback != sql.ErrTxDone {
+			err = multierror.Append(err, errRollback)
+		}
+		return errors.Wrap(ctx, err, op)
+	}
+	return nil
+}
+
+// Run will apply a migration. The io.Reader should provide the SQL
+// statements to execute, and the int is the version for that set of
+// statements. This should always be wrapped by StartRun and CommitRun.
+func (p *Postgres) Run(ctx context.Context, migration io.Reader, version int, edition string) error {
+	const op = "postgres.(Postgres).Run"
+
+	if p.tx == nil {
+		return errors.New(ctx, errors.MigrationIntegrity, op, "no pending transaction")
+	}
+
+	migr, err := ioutil.ReadAll(migration)
+	if err != nil {
+		return errors.Wrap(ctx, err, op)
+	}
+	// Run migration
+	query := string(migr)
+
+	// set the version first, so logs will be associated with this new version.
+	// if there's an error, it will get rollback
+	if err := p.setVersion(ctx, version, edition); err != nil {
+		return errors.Wrap(ctx, err, op)
+	}
+
+	if _, err := p.conn.ExecContext(ctx, query); err != nil {
+		if pgErr, ok := err.(*pgconn.PgError); ok {
+			var line uint
+			var col uint
+			var lineColOK bool
+			if pgErr.Position != 0 {
+				line, col, lineColOK = computeLineFromPos(query, int(pgErr.Position))
+			}
+			message := "migration failed"
+			if lineColOK {
+				message = fmt.Sprintf("%s (column %d)", message, col)
+			}
+			if pgErr.Detail != "" {
+				message = fmt.Sprintf("%s, %s", message, pgErr.Detail)
+			}
+			message = fmt.Sprintf("%s, on line %v: %s", message, line, migr)
+			return errors.Wrap(ctx, err, op, errors.WithMsg(message))
+		}
+		return errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("migration failed: %s", migr)))
+	}
+
+	return nil
+}
+
+var errOldMigrationTable = stderrors.New("old schema migration table")
+
+func (p *Postgres) schemaInitialized(ctx context.Context) (bool, error) {
+	const op = "postgres.(Postgres).CurrentState"
+	var initialized bool
+
+	tableNames := make([]string, 0, 2)
+	tableResult, err := p.conn.QueryContext(ctx, tablesExist)
+	if err != nil {
+		return false, errors.Wrap(ctx, err, op)
+	}
+	defer tableResult.Close()
+	for tableResult.Next() {
+		var tableName string
+		if err := tableResult.Scan(&tableName); err != nil {
+			return false, errors.Wrap(ctx, err, op)
+		}
+		tableNames = append(tableNames, tableName)
+	}
+
+	if len(tableNames) <= 0 {
+		// No version table found
+		return initialized, nil
+	}
+
+	initialized = true
+
+	if len(tableNames) > 1 {
+		return initialized, errors.New(ctx, errors.MigrationIntegrity, op, "both old and new migration tables exist")
+	}
+
+	if tableNames[0] != schemaVersionTable {
+		return initialized, errOldMigrationTable
+	}
+
+	return initialized, nil
+}
+
+// CurrentState returns the state of the given edition.
+// ver is the current migration version number as recorded in the database.
+// A version of -1 indicates no version is set.
+// initialized will be true if the schema was previously initialized.
+func (p *Postgres) CurrentState(ctx context.Context, edition string) (version int, initialized bool, err error) {
+	const op = "postgres.(Postgres).CurrentState"
+
+	version = nilVersion
+	initialized, err = p.schemaInitialized(ctx)
+	switch {
+	case err == errOldMigrationTable:
+		// only oss edition had the old migration table
+		if edition != "oss" {
+			return nilVersion, initialized, nil
+		}
+		err = p.conn.QueryRowContext(ctx, selectOldVersion).Scan(&version)
+		switch {
+		case err == sql.ErrNoRows:
+			// no version recorded
+			return nilVersion, initialized, nil
+		case err != nil:
+			return nilVersion, initialized, errors.Wrap(ctx, err, op)
+		default:
+			return version, initialized, nil
+		}
+	case err != nil:
+		return nilVersion, initialized, err
+	default:
+		// continue
+	}
+
+	if !initialized {
+		return nilVersion, initialized, nil
+	}
+
+	err = p.conn.QueryRowContext(ctx, selectVersion, edition).Scan(&version)
+	switch {
+	case err == sql.ErrNoRows:
+		// no version recorded
+		return nilVersion, initialized, nil
+	case err != nil:
+		// try to query for an edition in the pre-edition version table
+		if edition == "oss" {
+			preEditionErr := p.conn.QueryRowContext(ctx, selectPreEditionVersion).Scan(&version)
+			switch {
+			case preEditionErr == sql.ErrNoRows:
+				return nilVersion, initialized, nil
+			case preEditionErr != nil:
+				// return the original error
+				return nilVersion, initialized, errors.Wrap(ctx, err, op)
+			default:
+				return version, initialized, nil
+			}
+		}
+		return nilVersion, initialized, errors.Wrap(ctx, err, op)
+	default:
+		return version, initialized, nil
+	}
+}
+
+// EnsureVersionTable ensures that the table used to record the schema versions for each edition
+// exists and is in the correct state.
+func (p *Postgres) EnsureVersionTable(ctx context.Context) error {
+	const op = "postgres.(Postgres).EnsureVersionTable"
+
+	if p.tx == nil {
+		return errors.New(ctx, errors.MigrationIntegrity, op, "no pending transaction")
+	}
+
+	exists := false
+	if err := p.tx.QueryRowContext(ctx, tableExists).Scan(&exists); err != nil {
+		return errors.Wrap(ctx, err, op)
+	}
+
+	if !exists {
+		oldExists := false
+		if err := p.tx.QueryRowContext(ctx, oldTableExists).Scan(&oldExists); err != nil {
+			return errors.Wrap(ctx, err, op)
+		}
+		if !oldExists {
+			// create table for the first time it its correct state
+			if _, err := p.tx.ExecContext(ctx, migrations.Base("postgres").CreateSchemaVersion); err != nil {
+				return errors.Wrap(ctx, err, op)
+			}
+			return nil
+		}
+
+		// else convert old table to new table
+		if _, err := p.tx.ExecContext(ctx, updateTableName); err != nil {
+			return errors.Wrap(ctx, err, op)
+		}
+	}
+
+	// table exists, update it to ensure it has the correct columns and constraints
+	alterations := []string{
+		dropDirtyColumn,
+		addEditionColumn,
+		setVersionNotNull,
+	}
+
+	for _, a := range alterations {
+		if _, err := p.tx.ExecContext(ctx, a); err != nil {
+			return errors.Wrap(ctx, err, op)
+		}
+	}
+	return nil
+}
+
+// setVersion sets the version number, and whether the database is in a dirty state.
+// A version value of -1 indicates no version is set.
+func (p *Postgres) setVersion(ctx context.Context, version int, edition string) error {
+	const op = "postgres.(Postgres).setVersion"
+
+	if p.tx == nil {
+		return errors.New(ctx, errors.MigrationIntegrity, op, "no pending transaction")
+	}
+
+	if _, err := p.tx.ExecContext(ctx, upsertVersion, edition, version); err != nil {
+		return errors.Wrap(ctx, err, op)
+	}
+
+	return nil
+}
+
+// EnsureMigrationLogTable ensures that the table used to record migration lgos
+// exists and is in the correct state.
+func (p *Postgres) EnsureMigrationLogTable(ctx context.Context) error {
+	const op = "postgres.(Postgres).EnsureMigrationLogTable"
+
+	if p.tx == nil {
+		return errors.New(ctx, errors.MigrationIntegrity, op, "no pending transaction")
+	}
+
+	exists := false
+	if err := p.tx.QueryRowContext(ctx, migrationLogTableExists).Scan(&exists); err != nil {
+		return errors.Wrap(ctx, err, op)
+	}
+
+	if !exists {
+		if _, err := p.tx.ExecContext(ctx, migrations.Base("postgres").CreateLogMigration); err != nil {
+			return errors.Wrap(ctx, err, op)
+		}
+		return nil
+	}
+
+	// table exists, update it to ensure it has the correct columns and triggers
+	alterations := []string{
+		migrationLogAlterColumns,
+		migrationLogDropTriggers,
+		migrationLogAddEditionColumn,
+		migrationLogReplaceVersionTrigger,
+	}
+	for _, a := range alterations {
+		if _, err := p.tx.ExecContext(ctx, a); err != nil {
+			return errors.Wrap(ctx, err, op)
+		}
+	}
+	return nil
+}
+
+// GetMigrationLog will retrieve the migration logs from the db for the last
+// migration.
+//  The WithDeleteLog option is supported and will remove all log entries,
+// after reading the entries, when provided.
+func (p *Postgres) GetMigrationLog(ctx context.Context, opt ...log.Option) ([]*log.Entry, error) {
+	const op = "postgres.(Postgres).GetMigrationLog"
+
+	rows, err := p.conn.QueryContext(ctx, getMigrationLogs)
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op)
+	}
+	defer rows.Close()
+
+	var entries []*log.Entry
+	for rows.Next() {
+		e := &log.Entry{}
+
+		if err := rows.Scan(&e.Id, &e.CreateTime, &e.MigrationVersion, &e.MigrationEdition, &e.Entry); err != nil {
+			return nil, errors.Wrap(ctx, err, op)
+		}
+		entries = append(entries, e)
+	}
+	if rows.Err() != nil {
+		return nil, errors.Wrap(ctx, err, op)
+	}
+	opts := log.GetOpts(opt...)
+	if opts.WithDeleteLog {
+		// this truncate could change to a delete if FKs are needed in the future
+		_, err = p.conn.ExecContext(ctx, truncateMigrationLogs)
+		if err != nil {
+			return nil, errors.Wrap(ctx, err, op)
+		}
+	}
+	return entries, nil
+}

internal/db/schema/internal/postgres/postgres_test.go

@@ -0,0 +1,63 @@
+package postgres_test
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/errors"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRun(t *testing.T) {
+	ctx := context.Background()
+	p, _, _ := setup(ctx, t)
+
+	statements := bytes.NewReader([]byte(`
+create table foo (
+  id bigint primary key,
+  bar text
+);
+`))
+
+	err := p.StartRun(ctx)
+	require.NoError(t, err)
+
+	err = p.EnsureVersionTable(ctx)
+	require.NoError(t, err)
+
+	err = p.EnsureMigrationLogTable(ctx)
+	require.NoError(t, err)
+
+	err = p.Run(ctx, statements, 1001, "oss")
+	require.NoError(t, err)
+
+	err = p.CommitRun(ctx)
+	require.NoError(t, err)
+
+	v, i, err := p.CurrentState(ctx, "oss")
+	require.NoError(t, err)
+	assert.True(t, i)
+	assert.Equal(t, v, 1001)
+}
+
+func TestRun_NoTxn(t *testing.T) {
+	ctx := context.Background()
+	p, _, _ := setup(ctx, t)
+
+	statements := bytes.NewReader([]byte(`
+create table foo (
+  id bigint primary key,
+  bar text
+);
+`))
+
+	err := p.Run(ctx, statements, 1001, "oss")
+	require.EqualError(
+		t,
+		err,
+		fmt.Sprintf("postgres.(Postgres).Run: no pending transaction: integrity violation: error #%d", errors.MigrationIntegrity),
+	)
+}

internal/db/schema/internal/postgres/query.go

@@ -0,0 +1,139 @@
+package postgres
+
+// Queries for interacting with locks
+const (
+	trySharedLock    = `select pg_try_advisory_lock_shared($1)`
+	tryLock          = `select pg_try_advisory_lock($1)`
+	lock             = `select pg_advisory_lock($1)`
+	unlock           = `select pg_advisory_unlock($1)`
+	unlockSharedLock = `select pg_advisory_unlock_shared($1)`
+)
+
+// Queries for interacting with the schema version table
+const (
+	schemaVersionTable = `boundary_schema_version`
+	upsertVersion      = `
+insert into boundary_schema_version
+	(edition, version)
+values
+	($1, $2)
+on conflict (edition)
+do update set
+	version = $2
+;`
+
+	selectVersion = `
+select version
+  from boundary_schema_version
+ where edition = $1
+;`
+
+	selectOldVersion = `
+select version from schema_migrations;
+`
+
+	selectPreEditionVersion = `
+select max(version) from boundary_schema_version;
+`
+
+	tablesExist = `
+select table_name from information_schema.tables
+ where table_schema =  (select current_schema())
+   and table_name   in ('schema_migrations', 'boundary_schema_version')
+;`
+
+	tableExists = `
+select exists (
+	select 1 from information_schema.tables
+	 where table_schema = (select current_schema())
+	   and table_name   = 'boundary_schema_version'
+);`
+
+	oldTableExists = `
+select exists (
+	select 1 from information_schema.tables
+	 where table_schema = (select current_schema())
+	   and table_name   = 'schema_migrations'
+);`
+
+	updateTableName = `
+alter table if exists schema_migrations
+	rename to boundary_schema_version
+;`
+
+	dropDirtyColumn = `
+alter table boundary_schema_version
+	drop column if exists dirty
+;`
+
+	addEditionColumn = `
+alter table boundary_schema_version
+	add column if not exists
+	edition text default 'oss';
+alter table boundary_schema_version
+	alter column edition drop default;
+alter table boundary_schema_version drop constraint if exists schema_migrations_pkey;
+alter table boundary_schema_version drop constraint if exists boundary_schema_version_pkey;
+alter table boundary_schema_version add primary key (edition);
+`
+	setVersionNotNull = `
+alter table boundary_schema_version
+	alter column version set not null;
+`
+)
+
+// Quries for interacting with migration logs table
+const (
+	migrationLogTableExists = `
+select exists (
+	select 1 from information_schema.tables
+	 where table_schema = (select current_schema())
+	   and table_name   = 'log_migration'
+);`
+
+	getMigrationLogs = `
+select
+	id, create_time, migration_version, edition, entry
+ from log_migration
+where
+  migration_version in (
+    select max(version) from boundary_schema_version
+  );`
+
+	truncateMigrationLogs = `truncate log_migration;`
+
+	migrationLogAlterColumns = `
+alter table log_migration
+	alter column create_time type timestamp with time zone;
+alter table log_migration
+	alter column create_time set default current_timestamp;
+`
+	migrationLogDropTriggers = `
+drop trigger if exists default_create_time_column on log_migration;
+drop trigger if exists immutable_columns on log_migration;
+`
+	migrationLogAddEditionColumn = `
+alter table log_migration
+	add column if not exists
+	edition text default 'oss';
+alter table log_migration
+	alter column edition drop default;
+`
+	migrationLogReplaceVersionTrigger = `
+create or replace function
+  log_migration_version()
+  returns trigger
+as $$
+  declare current_version bigint;
+  begin
+    select max(version)
+	  from boundary_schema_version into current_version
+	 where edition = new.edition;
+    new.migration_version = current_version;
+    return new;
+  end;
+$$ language plpgsql;
+comment on function log_migration_version() is
+'log_migration_version will set the log_migration entries to the current migration version';
+`
+)

internal/db/schema/internal/postgres/state_test.go

@@ -0,0 +1,56 @@
+package postgres_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCurrentStatePreEdition(t *testing.T) {
+	ctx := context.Background()
+	p, db, _ := setup(ctx, t)
+
+	_, err := db.ExecContext(ctx, `
+	create table boundary_schema_version (
+		version bigint primary key,
+		dirty boolean not null
+	);`)
+	require.NoError(t, err)
+	_, err = db.ExecContext(ctx, `
+	insert into boundary_schema_version
+	(version, dirty)
+	values
+	(1001, false);`)
+	require.NoError(t, err)
+
+	version, initialized, err := p.CurrentState(ctx, "oss")
+	require.NoError(t, err)
+	assert.Equal(t, version, 1001)
+	assert.True(t, initialized)
+}
+
+func TestCurrentStateOldSchemaTable(t *testing.T) {
+	ctx := context.Background()
+	p, db, _ := setup(ctx, t)
+
+	_, err := db.ExecContext(ctx, `
+	create table schema_migrations (
+		version bigint primary key,
+		dirty boolean not null
+	);`)
+	require.NoError(t, err)
+	require.NoError(t, err)
+	_, err = db.ExecContext(ctx, `
+	insert into schema_migrations
+	(version, dirty)
+	values
+	(1001, false);`)
+	require.NoError(t, err)
+
+	version, initialized, err := p.CurrentState(ctx, "oss")
+	require.NoError(t, err)
+	assert.Equal(t, version, 1001)
+	assert.True(t, initialized)
+}

internal/db/schema/internal/postgres/testing_test.go

@@ -0,0 +1,35 @@
+package postgres_test
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/db/common"
+	"github.com/hashicorp/boundary/internal/db/schema/internal/postgres"
+	"github.com/hashicorp/boundary/testing/dbtest"
+	"github.com/stretchr/testify/require"
+)
+
+// setup is a helper function for tests.
+// It creates a new database using Template1, establishes an connection,
+// and returns a *postgres.Postgrs.
+// It also returns the underlying sql.DB and connection url so tests
+// can establish additional connections if necessary.
+func setup(ctx context.Context, t *testing.T) (*postgres.Postgres, *sql.DB, string) {
+	t.Helper()
+
+	c, u, _, err := dbtest.StartUsingTemplate(dbtest.Postgres, dbtest.WithTemplate(dbtest.Template1))
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, c())
+	})
+	d, err := common.SqlOpen(dbtest.Postgres, u)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, d.Close())
+	})
+	p, err := postgres.New(ctx, d)
+
+	return p, d, u
+}

internal/db/schema/internal/provider/provider.go

@@ -0,0 +1,100 @@
+// Package provider provides an iterator for iterating over all of the
+// migration statements that need to be applied. It will provide the statements
+// in the correct order based on the Edition priority and migration version.
+package provider
+
+import (
+	"sort"
+
+	"github.com/hashicorp/boundary/internal/db/schema/internal/edition"
+)
+
+const nilVersion = -1
+
+type migration struct {
+	version    int
+	edition    string
+	statements []byte
+}
+
+// Provider provides the migrations to the schema.Manager in the correct order.
+type Provider struct {
+	pos        int
+	migrations []migration
+}
+
+// DatabaseState is a map of edition names to versions.
+type DatabaseState map[string]int
+
+// New creates a Provider. The given DatabaseState is compared against the editions
+// to determine which migrations need to be applied and the migrations are then ordered
+// based on the Edition priority and by the migration version.
+func New(dbState DatabaseState, editions edition.Editions) *Provider {
+	m := &Provider{
+		pos: -1,
+	}
+
+	// ensure editions in priority order
+	editions.Sort()
+
+	allMigrations := make([]migration, 0)
+
+	for _, e := range editions {
+		dbVer, ok := dbState[e.Name]
+		if !ok {
+			dbVer = nilVersion
+		}
+
+		migrations := make([]migration, 0, len(e.Migrations))
+		for ver, statements := range e.Migrations {
+			if ver > dbVer {
+				migrations = append(migrations, migration{
+					version:    ver,
+					edition:    e.Name,
+					statements: statements,
+				})
+			}
+		}
+
+		sort.SliceStable(migrations, func(i, j int) bool {
+			return migrations[i].version < migrations[j].version
+		})
+
+		allMigrations = append(allMigrations, migrations...)
+	}
+
+	m.migrations = allMigrations
+
+	return m
+}
+
+// Next proceeds to the next migration. It returns true on success or false
+// if there are no more migrations.
+func (p *Provider) Next() bool {
+	p.pos++
+	return len(p.migrations) > p.pos
+}
+
+// Version returns the version for the current migration.
+func (p *Provider) Version() int {
+	if p.pos < 0 || p.pos >= len(p.migrations) {
+		return -1
+	}
+	return p.migrations[p.pos].version
+}
+
+// Edition returns the edition name for the current migration.
+func (p *Provider) Edition() string {
+	if p.pos < 0 || p.pos >= len(p.migrations) {
+		return ""
+	}
+	return p.migrations[p.pos].edition
+}
+
+// Statements returns the sql statements name for the current migration.
+func (p *Provider) Statements() []byte {
+	if p.pos < 0 || p.pos >= len(p.migrations) {
+		return nil
+	}
+	return p.migrations[p.pos].statements
+}

internal/db/schema/internal/provider/provider_test.go

@@ -0,0 +1,159 @@
+package provider_test
+
+import (
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/db/schema/internal/edition"
+	"github.com/hashicorp/boundary/internal/db/schema/internal/provider"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type expectedMigration struct {
+	version    int
+	edition    string
+	statements []byte
+}
+
+type expectedMigrations []expectedMigration
+
+func TestProvider(t *testing.T) {
+	tests := []struct {
+		name     string
+		editions edition.Editions
+		dbState  provider.DatabaseState
+		expected expectedMigrations
+	}{
+		{
+			"oneEditionNoneApplied",
+			edition.Editions{
+				edition.Edition{
+					Name:          "one",
+					LatestVersion: 2,
+					Migrations: map[int][]byte{
+						1: []byte(`migration one`),
+						2: []byte(`migration two`),
+					},
+					Priority: 0,
+				},
+			},
+			provider.DatabaseState{"one": -1},
+			expectedMigrations{
+				{1, "one", []byte(`migration one`)},
+				{2, "one", []byte(`migration two`)},
+			},
+		},
+		{
+			"twoEditionsNoneApplied",
+			edition.Editions{
+				edition.Edition{
+					Name:          "one",
+					LatestVersion: 2,
+					Migrations: map[int][]byte{
+						1: []byte(`migration one`),
+						2: []byte(`migration two`),
+					},
+					Priority: 0,
+				},
+				edition.Edition{
+					Name:          "two",
+					LatestVersion: 1,
+					Migrations: map[int][]byte{
+						1: []byte(`migration one`),
+					},
+					Priority: 0,
+				},
+			},
+			provider.DatabaseState{
+				"one": -1,
+			},
+			expectedMigrations{
+				{1, "one", []byte(`migration one`)},
+				{2, "one", []byte(`migration two`)},
+				{1, "two", []byte(`migration one`)},
+			},
+		},
+		{
+			"twoEditionsOnePartial",
+			edition.Editions{
+				edition.Edition{
+					Name:          "one",
+					LatestVersion: 2,
+					Migrations: map[int][]byte{
+						1: []byte(`migration one`),
+						2: []byte(`migration two`),
+					},
+					Priority: 0,
+				},
+				edition.Edition{
+					Name:          "two",
+					LatestVersion: 1,
+					Migrations: map[int][]byte{
+						1: []byte(`migration one`),
+					},
+					Priority: 0,
+				},
+			},
+			provider.DatabaseState{
+				"one": 1,
+				"two": -1,
+			},
+			expectedMigrations{
+				{2, "one", []byte(`migration two`)},
+				{1, "two", []byte(`migration one`)},
+			},
+		},
+		{
+			"twoEditionsBothPartial",
+			edition.Editions{
+				edition.Edition{
+					Name:          "one",
+					LatestVersion: 2,
+					Migrations: map[int][]byte{
+						1: []byte(`migration one`),
+						2: []byte(`migration two`),
+					},
+					Priority: 0,
+				},
+				edition.Edition{
+					Name:          "two",
+					LatestVersion: 1,
+					Migrations: map[int][]byte{
+						1: []byte(`migration one`),
+						2: []byte(`migration two`),
+					},
+					Priority: 0,
+				},
+			},
+			provider.DatabaseState{
+				"one": 1,
+				"two": 1,
+			},
+			expectedMigrations{
+				{2, "one", []byte(`migration two`)},
+				{2, "two", []byte(`migration two`)},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			p := provider.New(tt.dbState, tt.editions)
+
+			for _, expected := range tt.expected {
+				next := p.Next()
+				require.True(t, next)
+
+				assert.Equal(t, p.Version(), expected.version, tt.name)
+				assert.Equal(t, p.Edition(), expected.edition, tt.name)
+				assert.Equal(t, p.Statements(), expected.statements, tt.name)
+			}
+
+			assert.False(t, p.Next(), tt.name)
+			assert.Equal(t, p.Version(), -1, tt.name)
+			assert.Equal(t, p.Edition(), "", tt.name)
+			assert.Nil(t, p.Statements(), tt.name)
+		})
+	}
+
+}

internal/db/schema/manager.go

@@ -6,11 +6,12 @@ import (
 	"database/sql"
 	"fmt"
 	"io"
-	"time"
 
-	"github.com/hashicorp/boundary/internal/db/schema/postgres"
+	"github.com/hashicorp/boundary/internal/db/schema/internal/edition"
+	"github.com/hashicorp/boundary/internal/db/schema/internal/log"
+	"github.com/hashicorp/boundary/internal/db/schema/internal/postgres"
+	"github.com/hashicorp/boundary/internal/db/schema/internal/provider"
 	"github.com/hashicorp/boundary/internal/errors"
-	"github.com/hashicorp/go-multierror"
 )
 
 // driver provides functionality to a database.
@@ -20,43 +21,59 @@ type driver interface {
 	Lock(context.Context) error
 	Unlock(context.Context) error
 	UnlockShared(context.Context) error
-	// Either starts a transactioon internal to the driver or sets a dirty
-	// bit so if the Run fails the CurrentState reflects it.
+	// StartRun begins a transaction internal to the driver.
 	StartRun(context.Context) error
-	// Either commits the transaction or clears the dirty bit.
-	CommitRun() error
-	// Performs the mutation on the driver.  This should always be
-	// wrapped by StartRun and CommitRun.  The driver must properly
-	// handle the transaction or dirty bit in case of error when
-	// executing Run.
-	Run(context.Context, io.Reader, int) error
+	// CommitRun commits a transaction, if there is an error it should rollback the transaction.
+	CommitRun(context.Context) error
+	// Run will apply a migration. The io.Reader should provide the SQL
+	// statements to execute, and the int is the version for that set of
+	// statements. This should always be wrapped by StartRun and CommitRun.
+	Run(ctx context.Context, migration io.Reader, version int, edition string) error
+	// CurrentState returns the state of the given edition.
+	// ver is the current migration version number as recorded in the database.
 	// A version of -1 indicates no version is set.
-	CurrentState(context.Context) (ver int, everRan bool, dirty bool, err error)
+	// initialized will be true if the schema was previously initialized.
+	CurrentState(ctx context.Context, edition string) (version int, initialized bool, err error)
+	// EnsureVersionTable ensures that the table used to record the schema versions for each edition
+	// exists and is in the correct state.
 	EnsureVersionTable(ctx context.Context) error
+	// EnsureMigrationLogTable ensures that the table used to record migration lgos
+	// exists and is in the correct state.
+	EnsureMigrationLogTable(ctx context.Context) error
+	// GetMigrationLog will retrieve the migration logs from the db for the last
+	// migration.
+	//  The WithDeleteLog option is supported and will remove all log entries,
+	// after reading the entries, when provided.
+	GetMigrationLog(ctx context.Context, opt ...log.Option) ([]*log.Entry, error)
 }
 
 // Manager provides a way to run operations and retrieve information regarding
 // the underlying boundary database schema.
 // Manager is not thread safe.
 type Manager struct {
-	db              *sql.DB
-	driver          driver
-	dialect         string
-	migrationStates map[string]migrationState
+	db       *sql.DB
+	driver   driver
+	dialect  string
+	editions edition.Editions
 }
 
 // NewManager creates a new schema manager. An error is returned
 // if the provided dialect is unrecognized or if the passed in db is unreachable.
-func NewManager(ctx context.Context, dialect string, db *sql.DB, opt ...Option) (*Manager, error) {
+func NewManager(ctx context.Context, dialect Dialect, db *sql.DB, opt ...Option) (*Manager, error) {
 	const op = "schema.NewManager"
-	dbM := Manager{db: db, dialect: dialect}
+
+	editions.Lock()
+	defer editions.Unlock()
+
+	dbM := Manager{db: db, dialect: string(dialect)}
 	opts := getOpts(opt...)
-	if opts.withMigrationStates != nil {
-		dbM.migrationStates = opts.withMigrationStates
+	if opts.withEditions != nil {
+		dbM.editions = opts.withEditions
 	} else {
-		// intentionally set it to the reference, so changes to the global var
-		// will be reflected in the manager instance
-		dbM.migrationStates = migrationStates
+		dbM.editions = make(edition.Editions, 0, len(editions.m[edition.Dialect(dialect)]))
+		for _, e := range editions.m[edition.Dialect(dialect)] {
+			dbM.editions = append(dbM.editions, e)
+		}
 	}
 	switch dialect {
 	case "postgres":
@@ -71,32 +88,25 @@ func NewManager(ctx context.Context, dialect string, db *sql.DB, opt ...Option)
 	return &dbM, nil
 }
 
-// State contains information regarding the current state of a boundary database's schema.
-type State struct {
-	// InitializationStarted indicates if the current database has been initialized previously.
-	InitializationStarted bool
-	// Dirty is set to true if the database failed in a previous migration/initialization.
-	Dirty bool
-	// DatabaseSchemaVersion is the schema version that is currently running in the database.
-	DatabaseSchemaVersion int
-	// BinarySchemaVersion is the schema version which this boundary binary supports.
-	BinarySchemaVersion int
-}
-
 // CurrentState provides the state of the boundary schema contained in the backing database.
 func (b *Manager) CurrentState(ctx context.Context) (*State, error) {
 	const op = "schema.(Manager).CurrentState"
-	dbS := State{
-		BinarySchemaVersion: BinarySchemaVersion(b.dialect),
+	var dbS State
+
+	for _, e := range b.editions {
+		v, initialized, err := b.driver.CurrentState(ctx, e.Name)
+		if err != nil {
+			return nil, errors.Wrap(ctx, err, op)
+		}
+		dbS.Initialized = initialized || dbS.Initialized
+		dbS.Editions = append(dbS.Editions, EditionState{
+			Name:                  e.Name,
+			DatabaseSchemaVersion: v,
+			BinarySchemaVersion:   e.LatestVersion,
+			DatabaseSchemaState:   compareVersions(v, e.LatestVersion),
+		})
 	}
 
-	v, initialized, dirty, err := b.driver.CurrentState(ctx)
-	if err != nil {
-		return nil, errors.Wrap(ctx, err, op)
-	}
-	dbS.InitializationStarted = initialized
-	dbS.DatabaseSchemaVersion = v
-	dbS.Dirty = dirty
 	return &dbS, nil
 }
 
@@ -143,11 +153,11 @@ func (b *Manager) ExclusiveUnlock(ctx context.Context) error {
 	return nil
 }
 
-// RollForward updates the database schema to match the latest version known by
+// ApplyMigrations updates the database schema to match the latest version known by
 // the boundary binary.  An error is not returned if the database is already at
 // the most recent version.
-func (b *Manager) RollForward(ctx context.Context) error {
-	const op = "schema.(Manager).RollForward"
+func (b *Manager) ApplyMigrations(ctx context.Context) error {
+	const op = "schema.(Manager).ApplyMigrations"
 
 	// Capturing a lock that this session to the db already possesses is okay.
 	if err := b.driver.Lock(ctx); err != nil {
@@ -161,101 +171,57 @@ func (b *Manager) RollForward(ctx context.Context) error {
 		}
 	}()
 
-	curVersion, _, dirty, err := b.driver.CurrentState(ctx)
+	state, err := b.CurrentState(ctx)
 	if err != nil {
 		return errors.Wrap(ctx, err, op)
 	}
 
-	if dirty {
-		return errors.New(ctx, errors.NotSpecificIntegrity, op, fmt.Sprintf("schema is dirty with version %d", curVersion))
-	}
-
-	if err = b.runMigrations(ctx, newStatementProvider(b.dialect, curVersion, WithMigrationStates(b.migrationStates))); err != nil {
+	if err = b.runMigrations(ctx, provider.New(state.databaseState(), b.editions)); err != nil {
 		return errors.Wrap(ctx, err, op)
 	}
 	return nil
 }
 
-type rollbacker interface {
-	Rollback() error
-}
-
 // runMigrations passes migration queries to a database driver and manages
 // the version and dirty bit. Cancellation or deadline/timeout is managed
 // through the passed in context.
-func (b *Manager) runMigrations(ctx context.Context, qp *statementProvider) error {
+func (b *Manager) runMigrations(ctx context.Context, p *provider.Provider) (err error) {
 	const op = "schema.(Manager).runMigrations"
 
-	if err := b.driver.StartRun(ctx); err != nil {
-		return errors.Wrap(ctx, err, op)
+	if startErr := b.driver.StartRun(ctx); startErr != nil {
+		err = errors.Wrap(ctx, startErr, op)
+		return err
 	}
-	if err := b.driver.EnsureVersionTable(ctx); err != nil {
-		return errors.Wrap(ctx, err, op)
+
+	defer func() {
+		if commitErr := b.driver.CommitRun(ctx); commitErr != nil {
+			err = errors.Wrap(ctx, commitErr, op)
+		}
+	}()
+
+	if ensureErr := b.driver.EnsureVersionTable(ctx); ensureErr != nil {
+		err = errors.Wrap(ctx, ensureErr, op)
+		return err
 	}
-	for qp.Next() {
+
+	if ensureErr := b.driver.EnsureMigrationLogTable(ctx); ensureErr != nil {
+		err = errors.Wrap(ctx, ensureErr, op)
+		return err
+	}
+
+	for p.Next() {
 		select {
 		case <-ctx.Done():
-			err := ctx.Err()
-			if d, ok := b.driver.(rollbacker); ok {
-				if rbErr := d.Rollback(); rbErr != nil {
-					err = multierror.Append(err, rbErr)
-				}
-			}
-			return errors.Wrap(ctx, err, op)
+			err = errors.Wrap(ctx, ctx.Err(), op)
+			return err
 		default:
 			// context is not done yet. Continue on to the next query to execute.
 		}
-		if err := b.driver.Run(ctx, bytes.NewReader(qp.ReadUp()), qp.Version()); err != nil {
-			return errors.Wrap(ctx, err, op)
+		if runErr := b.driver.Run(ctx, bytes.NewReader(p.Statements()), p.Version(), p.Edition()); err != nil {
+			err = errors.Wrap(ctx, runErr, op)
+			return err
 		}
 	}
-	if err := b.driver.CommitRun(); err != nil {
-		return errors.Wrap(ctx, err, op)
-	}
+
 	return nil
 }
-
-// LogEntry represents a log entry generated during migrations.
-type LogEntry struct {
-	Id               int
-	MigrationVersion string
-	CreateTime       time.Time
-	Entry            string
-}
-
-// GetMigrationLog will retrieve the migration logs from the db for the last
-// migration. Once it's read the entries, it will delete them from the database.
-//  The WithDeleteLog option is supported and will remove all log entries when provided.
-func GetMigrationLog(ctx context.Context, d *sql.DB, opt ...Option) ([]LogEntry, error) {
-	const op = "schema.GetMigrationLog"
-	const sql = "select id, create_time, migration_version, entry from log_migration where migration_version in (select max(version) from boundary_schema_version)"
-	if d == nil {
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "missing sql db")
-	}
-	rows, err := d.QueryContext(ctx, sql)
-	if err != nil {
-		return nil, errors.Wrap(ctx, err, op)
-	}
-	defer rows.Close()
-
-	var entries []LogEntry
-	for rows.Next() {
-		var e LogEntry
-		if err := rows.Scan(&e.Id, &e.CreateTime, &e.MigrationVersion, &e.Entry); err != nil {
-			return nil, errors.Wrap(ctx, err, op)
-		}
-		entries = append(entries, e)
-	}
-	if rows.Err() != nil {
-		return nil, errors.Wrap(ctx, err, op)
-	}
-	opts := getOpts(opt...)
-	if opts.withDeleteLog {
-		// this truncate could change to a delete if FKs are needed in the future
-		_, err = d.ExecContext(ctx, "truncate log_migration")
-		if err != nil {
-			return nil, errors.Wrap(ctx, err, op)
-		}
-	}
-	return entries, nil
-}

internal/db/schema/manager_test.go

@@ -1,14 +1,14 @@
-package schema
+package schema_test
 
 import (
 	"context"
-	"database/sql"
 	"sort"
-	"strings"
 	"testing"
 
 	"github.com/hashicorp/boundary/internal/db/common"
-	"github.com/hashicorp/boundary/internal/db/schema/postgres"
+	"github.com/hashicorp/boundary/internal/db/schema"
+	"github.com/hashicorp/boundary/internal/db/schema/internal/edition"
+
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/testing/dbtest"
 	"github.com/stretchr/testify/assert"
@@ -27,13 +27,13 @@ func TestNewManager(t *testing.T) {
 	require.NoError(t, err)
 
 	ctx := context.Background()
-	_, err = NewManager(ctx, dialect, d)
+	_, err = schema.NewManager(ctx, schema.Dialect(dialect), d)
 	require.NoError(t, err)
-	_, err = NewManager(ctx, "unknown", d)
+	_, err = schema.NewManager(ctx, schema.Dialect("unknown"), d)
 	assert.True(t, errors.Match(errors.T(errors.InvalidParameter), err))
 
 	d.Close()
-	_, err = NewManager(ctx, dialect, d)
+	_, err = schema.NewManager(ctx, schema.Dialect(dialect), d)
 	assert.True(t, errors.Match(errors.T(errors.Op("schema.NewManager")), err))
 }
 
@@ -51,32 +51,232 @@ func TestCurrentState(t *testing.T) {
 	d, err := common.SqlOpen(dialect, u)
 	require.NoError(t, err)
 
-	m, err := NewManager(ctx, dialect, d)
+	m, err := schema.NewManager(ctx, schema.Dialect(dialect), d, schema.WithEditions(
+		edition.Editions{
+			{
+				Name:          "oss",
+				Dialect:       schema.Postgres,
+				LatestVersion: 2,
+				Migrations: map[int][]byte{
+					2: []byte(`select 1`),
+				},
+				Priority: 0,
+			},
+		},
+	))
 	require.NoError(t, err)
-	want := &State{
-		BinarySchemaVersion:   BinarySchemaVersion(dialect),
-		DatabaseSchemaVersion: nilVersion,
+	want := &schema.State{
+		Editions: []schema.EditionState{
+			{
+				Name:                  "oss",
+				BinarySchemaVersion:   2,
+				DatabaseSchemaVersion: schema.NilVersion,
+				DatabaseSchemaState:   schema.Behind,
+			},
+		},
 	}
 	s, err := m.CurrentState(ctx)
 	require.NoError(t, err)
 	assert.Equal(t, want, s)
 
-	testDriver, err := postgres.New(ctx, d)
-	require.NoError(t, err)
-	require.NoError(t, testDriver.EnsureVersionTable(ctx))
-	require.NoError(t, testDriver.Run(ctx, strings.NewReader("select 1"), 2))
+	assert.NoError(t, m.ApplyMigrations(ctx))
 
-	want = &State{
-		InitializationStarted: true,
-		BinarySchemaVersion:   BinarySchemaVersion(dialect),
-		DatabaseSchemaVersion: 2,
+	want = &schema.State{
+		Initialized: true,
+		Editions: []schema.EditionState{
+			{
+				Name:                  "oss",
+				BinarySchemaVersion:   2,
+				DatabaseSchemaVersion: 2,
+				DatabaseSchemaState:   schema.Equal,
+			},
+		},
 	}
 	s, err = m.CurrentState(ctx)
 	require.NoError(t, err)
 	assert.Equal(t, want, s)
+	assert.True(t, s.MigrationsApplied())
 }
 
-func TestRollForward(t *testing.T) {
+func TestApplyMigration(t *testing.T) {
+	tests := []struct {
+		name      string
+		editions  edition.Editions
+		expectErr bool
+		state     *schema.State
+	}{
+		{
+			"oneEdition",
+			edition.Editions{
+				edition.New("one", schema.Postgres, one, 0),
+			},
+			false,
+			&schema.State{
+				Initialized: true,
+				Editions: []schema.EditionState{
+					{
+						Name:                  "one",
+						BinarySchemaVersion:   1,
+						DatabaseSchemaVersion: 1,
+						DatabaseSchemaState:   schema.Equal,
+					},
+				},
+			},
+		},
+		{
+			"twoEditions",
+			edition.Editions{
+				edition.New("one", schema.Postgres, one, 0),
+				edition.New("two", schema.Postgres, two, 1),
+			},
+			false,
+			&schema.State{
+				Initialized: true,
+				Editions: []schema.EditionState{
+					{
+						Name:                  "one",
+						BinarySchemaVersion:   1,
+						DatabaseSchemaVersion: 1,
+						DatabaseSchemaState:   schema.Equal,
+					},
+					{
+						Name:                  "two",
+						BinarySchemaVersion:   1,
+						DatabaseSchemaVersion: 1,
+						DatabaseSchemaState:   schema.Equal,
+					},
+				},
+			},
+		},
+		{
+			"twoEditionsIncorrectPriority",
+			edition.Editions{
+				edition.New("one", schema.Postgres, one, 1),
+				edition.New("two", schema.Postgres, two, 0),
+			},
+			true,
+			&schema.State{
+				Initialized: false,
+				Editions: []schema.EditionState{
+					{
+						Name:                  "one",
+						BinarySchemaVersion:   1,
+						DatabaseSchemaVersion: schema.NilVersion,
+						DatabaseSchemaState:   schema.Behind,
+					},
+					{
+						Name:                  "two",
+						BinarySchemaVersion:   1,
+						DatabaseSchemaVersion: schema.NilVersion,
+						DatabaseSchemaState:   schema.Behind,
+					},
+				},
+			},
+		},
+		{
+			"threeEditions",
+			edition.Editions{
+				edition.New("one", schema.Postgres, one, 0),
+				edition.New("two", schema.Postgres, two, 1),
+				edition.New("three", schema.Postgres, three, 2),
+			},
+			false,
+			&schema.State{
+				Initialized: true,
+				Editions: []schema.EditionState{
+					{
+						Name:                  "one",
+						BinarySchemaVersion:   1,
+						DatabaseSchemaVersion: 1,
+						DatabaseSchemaState:   schema.Equal,
+					},
+					{
+						Name:                  "two",
+						BinarySchemaVersion:   1,
+						DatabaseSchemaVersion: 1,
+						DatabaseSchemaState:   schema.Equal,
+					},
+					{
+						Name:                  "three",
+						BinarySchemaVersion:   1,
+						DatabaseSchemaVersion: 1,
+						DatabaseSchemaState:   schema.Equal,
+					},
+				},
+			},
+		},
+		{
+			"threeEditionsIncorrectPriority",
+			edition.Editions{
+				edition.New("one", schema.Postgres, one, 0),
+				edition.New("two", schema.Postgres, two, 2),
+				edition.New("three", schema.Postgres, three, 1),
+			},
+			true,
+			&schema.State{
+				Initialized: false,
+				Editions: []schema.EditionState{
+					{
+						Name:                  "one",
+						BinarySchemaVersion:   1,
+						DatabaseSchemaVersion: schema.NilVersion,
+						DatabaseSchemaState:   schema.Behind,
+					},
+					{
+						Name:                  "two",
+						BinarySchemaVersion:   1,
+						DatabaseSchemaVersion: schema.NilVersion,
+						DatabaseSchemaState:   schema.Behind,
+					},
+					{
+						Name:                  "three",
+						BinarySchemaVersion:   1,
+						DatabaseSchemaVersion: schema.NilVersion,
+						DatabaseSchemaState:   schema.Behind,
+					},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			dialect := dbtest.Postgres
+
+			c, u, _, err := dbtest.StartUsingTemplate(dialect, dbtest.WithTemplate(dbtest.Template1))
+			t.Cleanup(func() {
+				if err := c(); err != nil {
+					t.Fatalf("Got error at cleanup: %v", err)
+				}
+			})
+			require.NoError(t, err)
+			ctx := context.Background()
+			d, err := common.SqlOpen(dialect, u)
+			require.NoError(t, err)
+
+			m, err := schema.NewManager(ctx, schema.Dialect(dialect), d, schema.WithEditions(tt.editions))
+			require.NoError(t, err)
+			if tt.expectErr {
+				assert.Error(t, m.ApplyMigrations(ctx))
+			} else {
+				assert.NoError(t, m.ApplyMigrations(ctx))
+			}
+
+			s, err := m.CurrentState(ctx)
+			require.NoError(t, err)
+			assert.Equal(t, tt.state.Initialized, s.Initialized)
+			assert.ElementsMatch(t, tt.state.Editions, s.Editions)
+
+			if tt.expectErr {
+				assert.False(t, s.MigrationsApplied())
+			} else {
+				assert.True(t, s.MigrationsApplied())
+			}
+		})
+	}
+}
+
+func TestApplyMigration_canceledContext(t *testing.T) {
 	dialect := dbtest.Postgres
 
 	c, u, _, err := dbtest.StartUsingTemplate(dialect)
@@ -88,87 +288,16 @@ func TestRollForward(t *testing.T) {
 	require.NoError(t, err)
 
 	ctx := context.Background()
-	m, err := NewManager(ctx, dialect, d)
-	require.NoError(t, err)
-	assert.NoError(t, m.RollForward(ctx))
-
-	// Now set to dirty at an early version
-	_, err = postgres.New(ctx, d)
-	require.NoError(t, err)
-	// TODO: Extract out a way to mock the db to test failing rollforwards.
-	_, err = d.ExecContext(ctx, "TRUNCATE boundary_schema_version; INSERT INTO boundary_schema_version (version, dirty) VALUES (2, true)")
-	require.NoError(t, err)
-	assert.Error(t, m.RollForward(ctx))
-}
-
-func TestRollForward_NotFromFresh(t *testing.T) {
-	dialect := dbtest.Postgres
-	oState := migrationStates[dialect]
-
-	nState := createPartialMigrationState(oState, 8)
-	migrationStates[dialect] = nState
-
-	c, u, _, err := dbtest.StartUsingTemplate(dialect, dbtest.WithTemplate(dbtest.Template1))
-	require.NoError(t, err)
-	t.Cleanup(func() {
-		require.NoError(t, c())
-	})
-	d, err := common.SqlOpen(dialect, u)
+	m, err := schema.NewManager(ctx, schema.Dialect(dialect), d)
 	require.NoError(t, err)
 
-	// Initialize the DB with only a portion of the current sql scripts.
-	ctx := context.Background()
-	m, err := NewManager(ctx, dialect, d)
-	require.NoError(t, err)
-	assert.NoError(t, m.RollForward(ctx))
-
-	state, err := m.CurrentState(ctx)
-	require.NoError(t, err)
-	assert.Equal(t, nState.binarySchemaVersion, state.DatabaseSchemaVersion)
-	assert.False(t, state.Dirty)
-
-	// Restore the full set of sql scripts and roll the rest of the way forward.
-	migrationStates[dialect] = oState
-
-	newM, err := NewManager(ctx, dialect, d)
-	require.NoError(t, err)
-	assert.NoError(t, newM.RollForward(ctx))
-	state, err = newM.CurrentState(ctx)
-	require.NoError(t, err)
-	assert.Equal(t, oState.binarySchemaVersion, state.DatabaseSchemaVersion)
-	assert.False(t, state.Dirty)
-}
-
-func TestRunMigration_canceledContext(t *testing.T) {
-	dialect := dbtest.Postgres
-
-	c, u, _, err := dbtest.StartUsingTemplate(dialect)
-	require.NoError(t, err)
-	t.Cleanup(func() {
-		require.NoError(t, c())
-	})
-	d, err := common.SqlOpen(dialect, u)
-	require.NoError(t, err)
-
-	ctx := context.Background()
-	m, err := NewManager(ctx, dialect, d)
-	require.NoError(t, err)
-
-	// TODO: Find a way to test different parts of the runMigrations loop.
 	ctx, cancel := context.WithCancel(ctx)
 	cancel()
-	assert.Error(t, m.runMigrations(ctx, newStatementProvider(dialect, 0)))
+	assert.Error(t, m.ApplyMigrations(ctx))
 }
 
-func TestRollForward_BadSQL(t *testing.T) {
+func TestApplyMigrations_BadSQL(t *testing.T) {
 	dialect := dbtest.Postgres
-	oState := migrationStates[dialect]
-	defer func() { migrationStates[dialect] = oState }()
-
-	nState := createPartialMigrationState(oState, 8)
-	nState.binarySchemaVersion = 10
-	nState.upMigrations[10] = []byte("SELECT 1 FROM NonExistantTable;")
-	migrationStates[dialect] = nState
 
 	c, u, _, err := dbtest.StartUsingTemplate(dialect, dbtest.WithTemplate(dbtest.Template1))
 	require.NoError(t, err)
@@ -180,14 +309,37 @@ func TestRollForward_BadSQL(t *testing.T) {
 
 	// Initialize the DB with only a portion of the current sql scripts.
 	ctx := context.Background()
-	m, err := NewManager(ctx, dialect, d)
+	m, err := schema.NewManager(ctx, schema.Dialect(dialect), d, schema.WithEditions(
+		edition.Editions{
+			{
+				Name:          "oss",
+				Dialect:       schema.Postgres,
+				LatestVersion: 1,
+				Migrations: map[int][]byte{
+					1: []byte(`select 1 from nonexistanttable;`),
+				},
+				Priority: 0,
+			},
+		},
+	))
 	require.NoError(t, err)
-	assert.Error(t, m.RollForward(ctx))
+	assert.Error(t, m.ApplyMigrations(ctx))
 
 	state, err := m.CurrentState(ctx)
 	require.NoError(t, err)
-	assert.Equal(t, nilVersion, state.DatabaseSchemaVersion)
-	assert.False(t, state.Dirty)
+	want := &schema.State{
+		Initialized: false,
+		Editions: []schema.EditionState{
+			{
+				Name:                  "oss",
+				BinarySchemaVersion:   1,
+				DatabaseSchemaVersion: schema.NilVersion,
+				DatabaseSchemaState:   schema.Behind,
+			},
+		},
+	}
+	assert.Equal(t, want, state)
+	assert.False(t, state.MigrationsApplied())
 }
 
 func TestManager_ExclusiveLock(t *testing.T) {
@@ -201,12 +353,12 @@ func TestManager_ExclusiveLock(t *testing.T) {
 	})
 	d1, err := common.SqlOpen(dialect, u)
 	require.NoError(t, err)
-	m1, err := NewManager(ctx, dialect, d1)
+	m1, err := schema.NewManager(ctx, schema.Dialect(dialect), d1)
 	require.NoError(t, err)
 
 	d2, err := common.SqlOpen(dialect, u)
 	require.NoError(t, err)
-	m2, err := NewManager(ctx, dialect, d2)
+	m2, err := schema.NewManager(ctx, schema.Dialect(dialect), d2)
 	require.NoError(t, err)
 
 	assert.NoError(t, m1.ExclusiveLock(ctx))
@@ -224,14 +376,14 @@ func TestManager_SharedLock(t *testing.T) {
 	t.Cleanup(func() {
 		require.NoError(t, c())
 	})
-	d1, err := common.SqlOpen("postgres", u)
+	d1, err := common.SqlOpen(dialect, u)
 	require.NoError(t, err)
-	m1, err := NewManager(ctx, "postgres", d1)
+	m1, err := schema.NewManager(ctx, schema.Dialect(dialect), d1)
 	require.NoError(t, err)
 
-	d2, err := common.SqlOpen("postgres", u)
+	d2, err := common.SqlOpen(dialect, u)
 	require.NoError(t, err)
-	m2, err := NewManager(ctx, "postgres", d2)
+	m2, err := schema.NewManager(ctx, schema.Dialect(dialect), d2)
 	require.NoError(t, err)
 
 	assert.NoError(t, m1.SharedLock(ctx))
@@ -253,22 +405,21 @@ func Test_GetMigrationLog(t *testing.T) {
 	t.Cleanup(func() {
 		require.NoError(t, c())
 	})
-	d, err := common.SqlOpen("postgres", u)
+	d, err := common.SqlOpen(dialect, u)
 	require.NoError(t, err)
-	m, err := NewManager(ctx, "postgres", d)
+	m, err := schema.NewManager(ctx, schema.Dialect(dialect), d)
 	require.NoError(t, err)
-	require.NoError(t, m.RollForward(ctx))
+	require.NoError(t, m.ApplyMigrations(ctx))
 
-	const insert = `insert into log_migration(entry) values ($1)`
+	const insert = `insert into log_migration(entry, edition) values ($1, $2)`
 	createEntries := func(entries ...string) {
 		for _, e := range entries {
-			_, err := d.Exec(insert, e)
+			_, err := d.Exec(insert, e, "oss")
 			require.NoError(t, err)
 		}
 	}
 	tests := []struct {
 		name          string
-		d             *sql.DB
 		setup         func()
 		withDeleteLog bool
 		wantEntries   []string
@@ -276,21 +427,15 @@ func Test_GetMigrationLog(t *testing.T) {
 	}{
 		{
 			name:        "simple",
-			d:           d,
 			setup:       func() { createEntries("alice", "eve", "bob") },
 			wantEntries: []string{"alice", "eve", "bob"},
 		},
 		{
 			name:          "with-delete-log",
-			d:             d,
 			setup:         func() { createEntries("alice", "eve", "bob") },
 			withDeleteLog: true,
 			wantEntries:   []string{"alice", "eve", "bob"},
 		},
-		{
-			name:         "missing-sql-DB",
-			wantErrMatch: errors.T(errors.InvalidParameter),
-		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -303,7 +448,7 @@ func Test_GetMigrationLog(t *testing.T) {
 			if tt.setup != nil {
 				tt.setup()
 			}
-			gotLog, err := GetMigrationLog(ctx, tt.d, WithDeleteLog(tt.withDeleteLog))
+			gotLog, err := m.GetMigrationLog(ctx, schema.WithDeleteLog(tt.withDeleteLog))
 			if tt.wantErrMatch != nil {
 				require.Error(err)
 				assert.Truef(errors.Match(tt.wantErrMatch, err), "expected error with code: %s and got err: %q", tt.wantErrMatch.Code, err)
@@ -330,21 +475,3 @@ func Test_GetMigrationLog(t *testing.T) {
 		})
 	}
 }
-
-// Creates a new migrationState only with the versions <= the provided maxVer
-func createPartialMigrationState(om migrationState, maxVer int) migrationState {
-	nState := migrationState{
-		upMigrations: make(map[int][]byte),
-	}
-	for k := range om.upMigrations {
-		if k > maxVer {
-			// Don't store any versions past our test version.
-			continue
-		}
-		nState.upMigrations[k] = om.upMigrations[k]
-		if nState.binarySchemaVersion < k {
-			nState.binarySchemaVersion = k
-		}
-	}
-	return nState
-}

internal/db/schema/migration_log.go

@@ -0,0 +1,49 @@
+package schema
+
+import (
+	"context"
+	"time"
+
+	"github.com/hashicorp/boundary/internal/db/schema/internal/log"
+	"github.com/hashicorp/boundary/internal/errors"
+)
+
+// LogEntry represents a log entry generated during migrations.
+type LogEntry struct {
+	Id               int
+	MigrationVersion int
+	MigrationEdition string
+	CreateTime       time.Time
+	Entry            string
+}
+
+// GetMigrationLog will retrieve the migration logs from the db for the last
+// migration. Once it's read the entries, it will delete them from the database.
+//  The WithDeleteLog option is supported and will remove all log entries when provided.
+func (b *Manager) GetMigrationLog(ctx context.Context, opt ...Option) ([]LogEntry, error) {
+	const op = "schema.GetMigrationLog"
+
+	var logOpts []log.Option
+	opts := getOpts(opt...)
+	if opts.withDeleteLog {
+		logOpts = append(logOpts, log.WithDeleteLog(opts.withDeleteLog))
+	}
+
+	entries, err := b.driver.GetMigrationLog(ctx, logOpts...)
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op)
+	}
+
+	logEntries := make([]LogEntry, 0, len(entries))
+	for _, e := range entries {
+		logEntries = append(logEntries, LogEntry{
+			Id:               e.Id,
+			MigrationVersion: e.MigrationVersion,
+			MigrationEdition: e.MigrationEdition,
+			CreateTime:       e.CreateTime,
+			Entry:            e.Entry,
+		})
+	}
+
+	return logEntries, nil
+}

internal/db/schema/migrations/README.md

@@ -1,27 +0,0 @@
-# migrations package
-This package handles the generation of the database schema in a format that can
-be compiled into the boundary binary.
-
-## Organization
-
-* `./generate`: contains the makefile, code, and templates needed to generate the schema info.
-* `./postgres`: contains the versioned schema folders.  The contents of these folders, except
- for `dev` should not be modified.
- 
-## Usage
-To regenerate the schema information into the format the boundary binary uses run
-`make migrations` or `make gen` to recreate all generated code.
-
-The content of the folders under `./postgres` are compiled into the
-boundary binary and when the `boundary database init` or `boundary database migrate`
-commands are executed they are applied in order of their version.
-
-The `./postgres/dev` directory contains schema files that are under development and
-are not included in a release yet and so it is the only directory where additions and
-modifications are allowed.  When a boundary binary is built when this directory is not 
-empty a special flag is required to run the `boundary database init` command to indicate
-the user is aware that this is a development release and running this command can
-result in a completely broken schema and dataloss.
-
-When a new release is made the contents of the `dev` directory are moved into a new
-versioned directory.

internal/db/schema/migrations/base.go

@@ -0,0 +1,66 @@
+// Package migrations contains the base sql statements needed to bootstrap the
+// migration process. These statements setup the tables necessary for tracking
+// the state of migrations.
+package migrations
+
+import (
+	"embed"
+	"path/filepath"
+	"strings"
+)
+
+// postgres contains the sql for creating the base tables used to manage
+// migrations.
+//go:embed base/postgres
+var postgres embed.FS
+
+// Base contains the sql migrtaions needed to create the initial based tables used
+// to manage other migrations.
+// Unlike the other migrations these create statements represent the "current" state of the tables
+// and are only used on a fresh install.
+type base struct {
+	CreateSchemaVersion string
+	CreateLogMigration  string
+}
+
+var baseByDialect map[string]base
+
+func stripBeginCommit(c []byte) string {
+	contents := strings.TrimSpace(string(c))
+	if strings.ToLower(contents[:len("begin;")]) == "begin;" {
+		contents = contents[len("begin;"):]
+	}
+	if strings.ToLower(contents[len(contents)-len("commit;"):]) == "commit;" {
+		contents = contents[:len(contents)-len("commit;")]
+	}
+	contents = strings.TrimSpace(contents)
+	return contents
+}
+
+func newBase(root string, m embed.FS) base {
+	schemaVersion, err := m.ReadFile(filepath.Join(root, "01_boundary_schema_version.up.sql"))
+	if err != nil {
+		panic("missing base migration file 01_boundary_schema_version.up.sql")
+	}
+
+	logMigration, err := m.ReadFile(filepath.Join(root, "02_log_migration.up.sql"))
+	if err != nil {
+		panic("missing base migration file 02_log_migration.up.sql")
+	}
+
+	return base{
+		CreateSchemaVersion: stripBeginCommit(schemaVersion),
+		CreateLogMigration:  stripBeginCommit(logMigration),
+	}
+}
+
+func init() {
+	baseByDialect = map[string]base{
+		"postgres": newBase("base/postgres", postgres),
+	}
+}
+
+// Base returns the base create statements for the given dialect.
+func Base(dialect string) base {
+	return baseByDialect[dialect]
+}

internal/db/schema/migrations/base/postgres/01_boundary_schema_version.up.sql

@@ -0,0 +1,6 @@
+begin;
+create table boundary_schema_version (
+	edition text   primary key,
+	version bigint not null
+);
+commit;

internal/db/schema/migrations/base/postgres/02_log_migration.up.sql

@@ -0,0 +1,34 @@
+begin;
+create table log_migration(
+  id                bigint generated always as identity primary key,
+  migration_version bigint not null, -- cannot declare FK since the table is truncated during runtime
+  edition           text   not null, -- cannot declare FK since the table is truncated during runtime
+  create_time       timestamp with time zone default current_timestamp,
+  entry             text   not null
+);
+comment on table log_migration is
+'log_migration entries are logging output from database migrations';
+
+create or replace function
+  log_migration_version()
+  returns trigger
+as $$
+  declare current_version bigint;
+  begin
+    select max(version)
+	  from boundary_schema_version into current_version
+	 where edition = new.edition;
+    new.migration_version = current_version;
+    return new;
+  end;
+$$ language plpgsql;
+comment on function log_migration_version() is
+'log_migration_version will set the log_migration entries to the current migration version';
+
+create trigger
+  version_column
+before
+  insert on log_migration
+  for each row execute procedure log_migration_version();
+
+commit;

internal/db/schema/migrations/generate/Makefile

@@ -1,9 +0,0 @@
-# Be sure to place this BEFORE `include` directives, if any.
-THIS_FILE := $(lastword $(MAKEFILE_LIST))
-
-migrations:
-	go run .
-	goimports -w ${GEN_BASEPATH}/internal/db/schema
-	gofumpt -w ${GEN_BASEPATH}/internal/db/schema
-
-.PHONY: migrations

internal/db/schema/migrations/generate/generate.go

@@ -1,145 +0,0 @@
-package main
-
-import (
-	"bytes"
-	"fmt"
-	"io/ioutil"
-	"os"
-	"sort"
-	"strconv"
-	"strings"
-	"text/template"
-)
-
-// generate looks for migration sql in a directory for the given dialect and
-// applies the templates below to the contents of the files, building up a
-// migrations map for the dialect
-func generate(dialect string) {
-	baseDir := os.Getenv("GEN_BASEPATH") + "/internal/db/schema"
-	srcDir := baseDir + "/migrations"
-	dir, err := os.Open(fmt.Sprintf("%s/%s", srcDir, dialect))
-	if err != nil {
-		fmt.Printf("error opening dir with dialect %s: %v\n", dialect, err)
-		os.Exit(1)
-	}
-	versions, err := dir.Readdirnames(0)
-	if err != nil {
-		fmt.Printf("error reading dir names with dialect %s: %v\n", dialect, err)
-		os.Exit(1)
-	}
-	sort.Strings(versions)
-
-	type ContentValues struct {
-		Name    string
-		Content string
-	}
-	var upContents []ContentValues
-
-	var largestSchemaVersion int
-	for _, ver := range versions {
-		if strings.HasPrefix(ver, ".") {
-			continue
-		}
-
-		var verVal int
-		verVal, err := strconv.Atoi(ver)
-		if err != nil {
-			fmt.Printf("error reading major schema version directory %q.  Must be a number: %v\n", ver, err)
-			os.Exit(1)
-		}
-
-		dir, err := os.Open(fmt.Sprintf("%s/%s/%s", srcDir, dialect, ver))
-		if err != nil {
-			fmt.Printf("error opening dir with dialect %s: %v\n", dialect, err)
-			os.Exit(1)
-		}
-		names, err := dir.Readdirnames(0)
-		if err != nil {
-			fmt.Printf("error reading dir names with dialect %s: %v\n", dialect, err)
-			os.Exit(1)
-		}
-
-		sort.Strings(names)
-		for _, name := range names {
-			if !strings.HasSuffix(name, ".sql") {
-				continue
-			}
-
-			nameParts := strings.SplitN(name, "_", 2)
-			if len(nameParts) != 2 {
-				continue
-			}
-
-			v, err := strconv.Atoi(nameParts[0])
-			if err != nil {
-				fmt.Printf("Unable to get file version from %q\n", name)
-				continue
-			}
-
-			fullV := (verVal * 1000) + v
-			if fullV > largestSchemaVersion {
-				largestSchemaVersion = fullV
-			}
-
-			cbts, err := ioutil.ReadFile(fmt.Sprintf("%s/%s/%s/%s", srcDir, dialect, ver, name))
-			if err != nil {
-				fmt.Printf("error opening file %s with dialect %s: %v", name, dialect, err)
-				os.Exit(1)
-			}
-
-			contents := strings.TrimSpace(string(cbts))
-			if strings.ToLower(contents[:len("begin;")]) == "begin;" {
-				contents = contents[len("begin;"):]
-			}
-			if strings.ToLower(contents[len(contents)-len("commit;"):]) == "commit;" {
-				contents = contents[:len(contents)-len("commit;")]
-			}
-			contents = strings.TrimSpace(contents)
-
-			cv := ContentValues{
-				Name:    fmt.Sprint(fullV),
-				Content: contents,
-			}
-			switch {
-			case strings.Contains(nameParts[1], ".up."):
-				upContents = append(upContents, cv)
-			}
-		}
-	}
-
-	outBuf := bytes.NewBuffer(nil)
-	if err := migrationsTemplate.Execute(outBuf, struct {
-		Type                string
-		UpValues            []ContentValues
-		BinarySchemaVersion int
-	}{
-		Type:                dialect,
-		UpValues:            upContents,
-		BinarySchemaVersion: largestSchemaVersion,
-	}); err != nil {
-		fmt.Printf("error executing migrations value template for dialect %s: %s", dialect, err)
-		os.Exit(1)
-	}
-
-	outFile := fmt.Sprintf("%s/%s_migration.gen.go", baseDir, dialect)
-	if err := ioutil.WriteFile(outFile, outBuf.Bytes(), 0o644); err != nil {
-		fmt.Printf("error writing file %q: %v\n", outFile, err)
-		os.Exit(1)
-	}
-}
-
-var migrationsTemplate = template.Must(template.Must(template.New("Content").Parse(
-	`{{ .Name }}: []byte(` + "`\n{{ .Content }}\n`" + `),
-`)).New("MainPage").Parse(`package schema
-
-// Code generated by "make migrations"; DO NOT EDIT.
-
-func init() {
-	migrationStates["{{ .Type }}"] = migrationState{
-		binarySchemaVersion: {{ .BinarySchemaVersion }},
-		upMigrations: map[int][]byte{
-			{{range .UpValues }}{{ template "Content" . }}{{end}}
-		},
-	}
-}
-`))

internal/db/schema/migrations/generate/main.go

@@ -1,5 +0,0 @@
-package main
-
-func main() {
-	generate("postgres")
-}

internal/db/schema/migrations/oss/oss.go

@@ -0,0 +1,17 @@
+// Package oss is used to embed the sql statements for the oss edition and
+// registering the edition for the schema.Manager.
+package oss
+
+import (
+	"embed"
+
+	"github.com/hashicorp/boundary/internal/db/schema"
+)
+
+// postgres contains the migrations sql files for postgres oss edition
+//go:embed postgres
+var postgres embed.FS
+
+func init() {
+	schema.RegisterEdition("oss", schema.Postgres, postgres, 0)
+}

internal/db/schema/migrations/oss/oss_test.go

@@ -0,0 +1,87 @@
+package oss_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/db/common"
+	"github.com/hashicorp/boundary/internal/db/schema"
+	"github.com/hashicorp/boundary/testing/dbtest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	// import for init side-effects to include migrations
+	_ "github.com/hashicorp/boundary/internal/db/schema/migrations/oss"
+)
+
+func TestApplyMigrations(t *testing.T) {
+	dialect := dbtest.Postgres
+
+	c, u, _, err := dbtest.StartUsingTemplate(dialect)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, c())
+	})
+	d, err := common.SqlOpen(dialect, u)
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	m, err := schema.NewManager(ctx, schema.Dialect(dialect), d)
+	require.NoError(t, err)
+	assert.NoError(t, m.ApplyMigrations(ctx))
+}
+
+func TestApplyMigrations_NotFromFresh(t *testing.T) {
+	dialect := dbtest.Postgres
+
+	c, u, _, err := dbtest.StartUsingTemplate(dialect, dbtest.WithTemplate(dbtest.Template1))
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, c())
+	})
+	d, err := common.SqlOpen(dialect, u)
+	require.NoError(t, err)
+
+	// Initialize the DB with only a portion of the current sql scripts.
+	ctx := context.Background()
+	m, err := schema.NewManager(ctx, schema.Dialect(dialect), d, schema.WithEditions(
+		schema.TestCreatePartialEditions(schema.Dialect(dialect), schema.PartialEditions{"oss": 1}),
+	))
+	require.NoError(t, err)
+	assert.NoError(t, m.ApplyMigrations(ctx))
+
+	state, err := m.CurrentState(ctx)
+	require.NoError(t, err)
+	want := &schema.State{
+		Initialized: true,
+		Editions: []schema.EditionState{
+			{
+				Name:                  "oss",
+				BinarySchemaVersion:   1,
+				DatabaseSchemaVersion: 1,
+				DatabaseSchemaState:   schema.Equal,
+			},
+		},
+	}
+	assert.Equal(t, want, state)
+
+	newM, err := schema.NewManager(ctx, schema.Dialect(dialect), d, schema.WithEditions(
+		schema.TestCreatePartialEditions(schema.Dialect(dialect), schema.PartialEditions{"oss": 3}),
+	))
+	require.NoError(t, err)
+	assert.NoError(t, newM.ApplyMigrations(ctx))
+	state, err = newM.CurrentState(ctx)
+	require.NoError(t, err)
+	want = &schema.State{
+		Initialized: true,
+		Editions: []schema.EditionState{
+			{
+				Name:                  "oss",
+				BinarySchemaVersion:   3,
+				DatabaseSchemaVersion: 3,
+				DatabaseSchemaState:   schema.Equal,
+			},
+		},
+	}
+	assert.Equal(t, want, state)
+}

internal/db/schema/migrations/oss/postgres/2/01_log.up.sql

@@ -0,0 +1,54 @@
+begin;
+
+-- management of the log_migation table is moved into
+-- the schema.Manager prior to migrations running
+-- since it interacts with the boundary_schema_version table
+
+-- log_migration entries represent logs generated during migrations
+-- create table log_migration(
+--   id bigint generated always as identity primary key,
+--   migration_version bigint not null, -- cannot declare FK since the table is truncated during runtime
+--   create_time wt_timestamp,
+--   entry text not null
+-- );
+-- comment on table log_migration is 
+-- 'log_migration entries are logging output from databaes migrations';
+-- 
+-- -- log_migration triggers
+-- create trigger 
+--   default_create_time_column
+-- before
+-- insert on log_migration
+--   for each row execute procedure default_create_time();
+-- 
+-- create trigger
+--   immutable_columns
+-- before
+-- update on log_migration
+--   for each row execute procedure immutable_columns('id', 'migration_version', 'create_time', 'entry');
+-- 
+-- 
+-- -- log_migration_version() defines a function to be used in a "before update"
+-- -- trigger for log_migrations entries.  Its intent: set the log_migration
+-- -- version column to the current migration version.  
+-- create or replace function
+--   log_migration_version()
+--   returns trigger
+-- as $$
+--   declare current_version bigint;
+--   begin
+--     select max(version) from boundary_schema_version into current_version;
+--     new.migration_version = current_version;
+--     return new;
+--   end;
+-- $$ language plpgsql;
+-- comment on function log_migration_version() is
+-- 'log_migration_version will set the log_migration entries to the current migration version';    
+-- 
+-- create trigger
+--   migration_version_column
+-- before
+-- insert on log_migration
+--   for each row execute procedure log_migration_version();
+-- 
+commit;