Add migration that brings line column to the migration table

CHANGELOG.md

@@ -7,10 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+⚠️ Version 0.10.0 contains a new database migration, version 5. See [documentation on running River migrations](https://riverqueue.com/docs/migrations). If migrating with the CLI, make sure to update it to its latest version:
+
+```shell
+go install github.com/riverqueue/river/cmd/river@latest
+river migrate-up --database-url "$DATABASE_URL"
+```
+
 ### Added
 
 - Fully functional driver for `database/sql` for use with packages like Bun and GORM. [PR #351](https://github.com/riverqueue/river/pull/351).
 - Queues can be added after a client is initialized using `client.Queues().Add(queueName string, queueConfig QueueConfig)`. [PR #410](https://github.com/riverqueue/river/pull/410).
+- Migration that adds a `line` column to the `river_migration` table so that it can support multiple migration lines. [PR #435](https://github.com/riverqueue/river/pull/435).
 
 ### Changed
 

cmd/river/go.mod

@@ -1,22 +1,22 @@
 module github.com/riverqueue/river/cmd/river
 
-go 1.21.4
+go 1.22.5
 
-// replace github.com/riverqueue/river => ../..
+replace github.com/riverqueue/river => ../..
 
-// replace github.com/riverqueue/river/riverdriver => ../../riverdriver
+replace github.com/riverqueue/river/riverdriver => ../../riverdriver
 
-// replace github.com/riverqueue/river/riverdriver/riverdatabasesql => ../../riverdriver/riverdatabasesql
+replace github.com/riverqueue/river/riverdriver/riverdatabasesql => ../../riverdriver/riverdatabasesql
 
-// replace github.com/riverqueue/river/riverdriver/riverpgxv5 => ../../riverdriver/riverpgxv5
+replace github.com/riverqueue/river/riverdriver/riverpgxv5 => ../../riverdriver/riverpgxv5
 
 require (
-	github.com/jackc/pgx/v5 v5.5.5
+	github.com/jackc/pgx/v5 v5.6.0
 	github.com/lmittmann/tint v1.0.4
 	github.com/riverqueue/river v0.6.1
-	github.com/riverqueue/river/riverdriver v0.6.1
-	github.com/riverqueue/river/riverdriver/riverpgxv5 v0.6.1
-	github.com/riverqueue/river/rivertype v0.6.1
+	github.com/riverqueue/river/riverdriver v0.9.0
+	github.com/riverqueue/river/riverdriver/riverpgxv5 v0.9.0
+	github.com/riverqueue/river/rivertype v0.9.0
 	github.com/spf13/cobra v1.8.0
 	github.com/stretchr/testify v1.9.0
 )
@@ -28,9 +28,11 @@ require (
 	github.com/jackc/pgservicefile v0.0.0-20231201235250-de7065d80cb9 // indirect
 	github.com/jackc/puddle/v2 v2.2.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/riverqueue/river/rivershared v0.0.0-20240707210043-f9063791ecb1 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
+	go.uber.org/goleak v1.3.0 // indirect
 	golang.org/x/crypto v0.23.0 // indirect
 	golang.org/x/sync v0.7.0 // indirect
-	golang.org/x/text v0.15.0 // indirect
+	golang.org/x/text v0.16.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

cmd/river/go.sum

@@ -10,8 +10,8 @@ github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsI
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
 github.com/jackc/pgservicefile v0.0.0-20231201235250-de7065d80cb9 h1:L0QtFUgDarD7Fpv9jeVMgy/+Ec0mtnmYuImjTz6dtDA=
 github.com/jackc/pgservicefile v0.0.0-20231201235250-de7065d80cb9/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
-github.com/jackc/pgx/v5 v5.5.5 h1:amBjrZVmksIdNjxGW/IiIMzxMKZFelXbUoPNb+8sjQw=
-github.com/jackc/pgx/v5 v5.5.5/go.mod h1:ez9gk+OAat140fv9ErkZDYFWmXLfV+++K0uAOiwgm1A=
+github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY=
+github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw=
 github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
 github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=
@@ -24,20 +24,14 @@ github.com/lmittmann/tint v1.0.4 h1:LeYihpJ9hyGvE0w+K2okPTGUdVLfng1+nDNVR4vWISc=
 github.com/lmittmann/tint v1.0.4/go.mod h1:HIS3gSy7qNwGCj+5oRjAutErFBl4BzdQP6cJZ0NfMwE=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/riverqueue/river v0.6.1 h1:D0A139oRJh3EvATSuMPwPVscoujdezxf32Osqf/ym9Q=
-github.com/riverqueue/river v0.6.1/go.mod h1:saKYj0h5bwon4377dMF5Lnq0yxtYLerjssuhJESY/nI=
-github.com/riverqueue/river/riverdriver v0.6.1 h1:xIgMgqC4WPpFT3yddvfpeeO7O2xJogBugOZVkDbVpjM=
-github.com/riverqueue/river/riverdriver v0.6.1/go.mod h1:l1QlqYXt+CFnVuS9ovNIvntnTeUeHuuDvEb/utvWO9k=
-github.com/riverqueue/river/riverdriver/riverdatabasesql v0.6.1 h1:ahJ9gpWEBxHazwRGiwxVbG1kVd3ceeMzq4eiZvW1C1E=
-github.com/riverqueue/river/riverdriver/riverdatabasesql v0.6.1/go.mod h1:58F2aScxSs4V9uxjL02kdzGeTr+ZviYdrSM9bLBC+sc=
-github.com/riverqueue/river/riverdriver/riverpgxv5 v0.6.1 h1:iHy7r+AHjSMBNRTVox/8K442zxhwfQhRqXPxqNpmJsY=
-github.com/riverqueue/river/riverdriver/riverpgxv5 v0.6.1/go.mod h1:XBxscxZzQCkempAXvBUwkXsh+DinBRIWnPBUNTUYd14=
-github.com/riverqueue/river/rivertype v0.6.1 h1:vW64T/sJN//5XkI2tNhNwK9nDfM4lobCm3d2AZyrQ70=
-github.com/riverqueue/river/rivertype v0.6.1/go.mod h1:nDd50b/mIdxR/ezQzGS/JiAhBPERA7tUIne21GdfspQ=
+github.com/riverqueue/river/rivershared v0.0.0-20240707210043-f9063791ecb1 h1:wCAWAmchE27xz40FBxmcoHxRWXzhiC94GecbguFk2S4=
+github.com/riverqueue/river/rivershared v0.0.0-20240707210043-f9063791ecb1/go.mod h1:2egnQ7czNcW8IXKXMRjko0aEMrQzF4V3k3jddmYiihE=
+github.com/riverqueue/river/rivertype v0.9.0 h1:xr2ktQ55lqqKgXIm0Z7GJDtGuKk9BUD9kbchoUL69Lg=
+github.com/riverqueue/river/rivertype v0.9.0/go.mod h1:nDd50b/mIdxR/ezQzGS/JiAhBPERA7tUIne21GdfspQ=
 github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
 github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
-github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
-github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
+github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
+github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/spf13/cobra v1.8.0 h1:7aJaZx1B85qltLMc546zn58BxxfZdR/W22ej9CFoEf0=
 github.com/spf13/cobra v1.8.0/go.mod h1:WXLWApfZ71AjXPya3WOlMsY9yMs7YeiHhFVlvLyhcho=
@@ -54,8 +48,8 @@ golang.org/x/crypto v0.23.0 h1:dIJU/v2J8Mdglj/8rJ6UUOM3Zc9zLZxVZwwxMooUSAI=
 golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8=
 golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
 golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
-golang.org/x/text v0.15.0 h1:h1V/4gjBv8v9cjcR6+AR5+/cIYK5N/WAgiv4xlsEtAk=
-golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/text v0.16.0 h1:a94ExnEXNtEwYLGJSIUxnWoxoRz/ZcCsV63ROupILh4=
+golang.org/x/text v0.16.0/go.mod h1:GhwF1Be+LQoKShO3cGOHzqOgRrGaYc9AvblQOmPVHnI=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=

go.mod

@@ -19,7 +19,7 @@ require (
 	github.com/riverqueue/river/riverdriver v0.9.0
 	github.com/riverqueue/river/riverdriver/riverdatabasesql v0.9.0
 	github.com/riverqueue/river/riverdriver/riverpgxv5 v0.9.0
-	github.com/riverqueue/river/rivershared v0.0.0-20240707170519-d0685f5e0a5d
+	github.com/riverqueue/river/rivershared v0.0.0-20240707210043-f9063791ecb1
 	github.com/riverqueue/river/rivertype v0.9.0
 	github.com/robfig/cron/v3 v3.0.1
 	github.com/stretchr/testify v1.9.0

internal/riverinternaltest/riverdrivertest/riverdrivertest.go

@@ -180,6 +180,20 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 		})
 	})
 
+	t.Run("ColumnExists", func(t *testing.T) {
+		t.Parallel()
+
+		exec, _ := setup(ctx, t)
+
+		exists, err := exec.ColumnExists(ctx, "river_job", "id")
+		require.NoError(t, err)
+		require.True(t, exists)
+
+		exists, err = exec.ColumnExists(ctx, "river_job", "does_not_exist")
+		require.NoError(t, err)
+		require.False(t, exists)
+	})
+
 	t.Run("Exec", func(t *testing.T) {
 		t.Parallel()
 
@@ -1877,7 +1891,7 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 		require.NoError(t, err)
 	}
 
-	t.Run("MigrationDeleteByVersionMany", func(t *testing.T) {
+	t.Run("MigrationDeleteAssumingMainMany", func(t *testing.T) {
 		t.Parallel()
 
 		exec, _ := setup(ctx, t)
@@ -1887,18 +1901,53 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 		migration1 := testfactory.Migration(ctx, t, exec, &testfactory.MigrationOpts{})
 		migration2 := testfactory.Migration(ctx, t, exec, &testfactory.MigrationOpts{})
 
-		migrations, err := exec.MigrationDeleteByVersionMany(ctx, []int{
+		// This query is designed to work before the `line` column was added to
+		// the `river_migration` table. These tests will be operating on a fully
+		// migrated database, so drop the column in this transaction to make
+		// sure we are really checking that this operation works as expected.
+		_, err := exec.Exec(ctx, "ALTER TABLE river_migration DROP COLUMN line")
+		require.NoError(t, err)
+
+		migrations, err := exec.MigrationDeleteAssumingMainMany(ctx, []int{
+			migration1.Version,
+			migration2.Version,
+		})
+		require.NoError(t, err)
+		require.Len(t, migrations, 2)
+		slices.SortFunc(migrations, func(a, b *riverdriver.Migration) int { return a.Version - b.Version })
+		require.Equal(t, riverdriver.MigrationLineMain, migrations[0].Line)
+		require.Equal(t, migration1.Version, migrations[0].Version)
+		require.Equal(t, riverdriver.MigrationLineMain, migrations[1].Line)
+		require.Equal(t, migration2.Version, migrations[1].Version)
+	})
+
+	t.Run("MigrationDeleteByLineAndVersionMany", func(t *testing.T) {
+		t.Parallel()
+
+		exec, _ := setup(ctx, t)
+
+		truncateMigrations(ctx, t, exec)
+
+		// not touched
+		_ = testfactory.Migration(ctx, t, exec, &testfactory.MigrationOpts{})
+
+		migration1 := testfactory.Migration(ctx, t, exec, &testfactory.MigrationOpts{Line: ptrutil.Ptr("alternate")})
+		migration2 := testfactory.Migration(ctx, t, exec, &testfactory.MigrationOpts{Line: ptrutil.Ptr("alternate")})
+
+		migrations, err := exec.MigrationDeleteByLineAndVersionMany(ctx, "alternate", []int{
 			migration1.Version,
 			migration2.Version,
 		})
 		require.NoError(t, err)
 		require.Len(t, migrations, 2)
 		slices.SortFunc(migrations, func(a, b *riverdriver.Migration) int { return a.Version - b.Version })
+		require.Equal(t, "alternate", migrations[0].Line)
 		require.Equal(t, migration1.Version, migrations[0].Version)
+		require.Equal(t, "alternate", migrations[1].Line)
 		require.Equal(t, migration2.Version, migrations[1].Version)
 	})
 
-	t.Run("MigrationGetAll", func(t *testing.T) {
+	t.Run("MigrationGetAllAssumingMain", func(t *testing.T) {
 		t.Parallel()
 
 		exec, _ := setup(ctx, t)
@@ -1908,7 +1957,14 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 		migration1 := testfactory.Migration(ctx, t, exec, &testfactory.MigrationOpts{})
 		migration2 := testfactory.Migration(ctx, t, exec, &testfactory.MigrationOpts{})
 
-		migrations, err := exec.MigrationGetAll(ctx)
+		// This query is designed to work before the `line` column was added to
+		// the `river_migration` table. These tests will be operating on a fully
+		// migrated database, so drop the column in this transaction to make
+		// sure we are really checking that this operation works as expected.
+		_, err := exec.Exec(ctx, "ALTER TABLE river_migration DROP COLUMN line")
+		require.NoError(t, err)
+
+		migrations, err := exec.MigrationGetAllAssumingMain(ctx)
 		require.NoError(t, err)
 		require.Len(t, migrations, 2)
 		require.Equal(t, migration1.Version, migrations[0].Version)
@@ -1918,6 +1974,34 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 		migration1Fetched := migrations[0]
 		require.Equal(t, migration1.ID, migration1Fetched.ID)
 		requireEqualTime(t, migration1.CreatedAt, migration1Fetched.CreatedAt)
+		require.Equal(t, riverdriver.MigrationLineMain, migration1Fetched.Line)
+		require.Equal(t, migration1.Version, migration1Fetched.Version)
+	})
+
+	t.Run("MigrationGetByLine", func(t *testing.T) {
+		t.Parallel()
+
+		exec, _ := setup(ctx, t)
+
+		truncateMigrations(ctx, t, exec)
+
+		// not returned
+		_ = testfactory.Migration(ctx, t, exec, &testfactory.MigrationOpts{})
+
+		migration1 := testfactory.Migration(ctx, t, exec, &testfactory.MigrationOpts{Line: ptrutil.Ptr("alternate")})
+		migration2 := testfactory.Migration(ctx, t, exec, &testfactory.MigrationOpts{Line: ptrutil.Ptr("alternate")})
+
+		migrations, err := exec.MigrationGetByLine(ctx, "alternate")
+		require.NoError(t, err)
+		require.Len(t, migrations, 2)
+		require.Equal(t, migration1.Version, migrations[0].Version)
+		require.Equal(t, migration2.Version, migrations[1].Version)
+
+		// Check the full properties of one of the migrations.
+		migration1Fetched := migrations[0]
+		require.Equal(t, migration1.ID, migration1Fetched.ID)
+		requireEqualTime(t, migration1.CreatedAt, migration1Fetched.CreatedAt)
+		require.Equal(t, "alternate", migration1Fetched.Line)
 		require.Equal(t, migration1.Version, migration1Fetched.Version)
 	})
 
@@ -1928,10 +2012,35 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 
 		truncateMigrations(ctx, t, exec)
 
-		migrations, err := exec.MigrationInsertMany(ctx, []int{1, 2})
+		migrations, err := exec.MigrationInsertMany(ctx, "alternate", []int{1, 2})
+		require.NoError(t, err)
+		require.Len(t, migrations, 2)
+		require.Equal(t, "alternate", migrations[0].Line)
+		require.Equal(t, 1, migrations[0].Version)
+		require.Equal(t, "alternate", migrations[1].Line)
+		require.Equal(t, 2, migrations[1].Version)
+	})
+
+	t.Run("MigrationInsertManyAssumingMain", func(t *testing.T) {
+		t.Parallel()
+
+		exec, _ := setup(ctx, t)
+
+		truncateMigrations(ctx, t, exec)
+
+		// This query is designed to work before the `line` column was added to
+		// the `river_migration` table. These tests will be operating on a fully
+		// migrated database, so drop the column in this transaction to make
+		// sure we are really checking that this operation works as expected.
+		_, err := exec.Exec(ctx, "ALTER TABLE river_migration DROP COLUMN line")
+		require.NoError(t, err)
+
+		migrations, err := exec.MigrationInsertManyAssumingMain(ctx, []int{1, 2})
 		require.NoError(t, err)
 		require.Len(t, migrations, 2)
+		require.Equal(t, riverdriver.MigrationLineMain, migrations[0].Line)
 		require.Equal(t, 1, migrations[0].Version)
+		require.Equal(t, riverdriver.MigrationLineMain, migrations[1].Line)
 		require.Equal(t, 2, migrations[1].Version)
 	})
 

internal/riverinternaltest/testfactory/test_factory.go

@@ -99,15 +99,17 @@ func Leader(ctx context.Context, tb testing.TB, exec riverdriver.Executor, opts
 }
 
 type MigrationOpts struct {
+	Line    *string
 	Version *int
 }
 
 func Migration(ctx context.Context, tb testing.TB, exec riverdriver.Executor, opts *MigrationOpts) *riverdriver.Migration {
 	tb.Helper()
 
-	migration, err := exec.MigrationInsertMany(ctx, []int{
-		ptrutil.ValOrDefaultFunc(opts.Version, nextSeq),
-	})
+	migration, err := exec.MigrationInsertMany(ctx,
+		ptrutil.ValOrDefault(opts.Line, riverdriver.MigrationLineMain),
+		[]int{ptrutil.ValOrDefaultFunc(opts.Version, nextSeq)},
+	)
 	require.NoError(tb, err)
 	return migration[0]
 }

riverdriver/river_driver_interface.go

@@ -98,6 +98,10 @@ type Executor interface {
 	// subtransactions (like riverdriver/riverdatabasesql for database/sql).
 	Begin(ctx context.Context) (ExecutorTx, error)
 
+	// ColumnExists checks whether a column for a particular table exists for
+	// the schema in the current search schema.
+	ColumnExists(ctx context.Context, tableName, columnName string) (bool, error)
+
 	// Exec executes raw SQL. Used for migrations.
 	Exec(ctx context.Context, sql string) (struct{}, error)
 
@@ -129,14 +133,30 @@ type Executor interface {
 	LeaderInsert(ctx context.Context, params *LeaderInsertParams) (*Leader, error)
 	LeaderResign(ctx context.Context, params *LeaderResignParams) (bool, error)
 
-	// MigrationDeleteByVersionMany deletes many migration versions.
-	MigrationDeleteByVersionMany(ctx context.Context, versions []int) ([]*Migration, error)
+	// MigrationDeleteAssumingMainMany deletes many migrations assuming
+	// everything is on the main line. This is suitable for use in databases on
+	// a version before the `line` column exists.
+	MigrationDeleteAssumingMainMany(ctx context.Context, versions []int) ([]*Migration, error)
+
+	// MigrationDeleteByLineAndVersionMany deletes many migration versions on a
+	// particular line.
+	MigrationDeleteByLineAndVersionMany(ctx context.Context, line string, versions []int) ([]*Migration, error)
 
-	// MigrationGetAll gets all currently applied migrations.
-	MigrationGetAll(ctx context.Context) ([]*Migration, error)
+	// MigrationGetAllAssumingMain gets all migrations assuming everything is on
+	// the main line. This is suitable for use in databases on a version before
+	// the `line` column exists.
+	MigrationGetAllAssumingMain(ctx context.Context) ([]*Migration, error)
+
+	// MigrationGetByLine gets all currently applied migrations.
+	MigrationGetByLine(ctx context.Context, line string) ([]*Migration, error)
 
 	// MigrationInsertMany inserts many migration versions.
-	MigrationInsertMany(ctx context.Context, versions []int) ([]*Migration, error)
+	MigrationInsertMany(ctx context.Context, line string, versions []int) ([]*Migration, error)
+
+	// MigrationInsertManyAssumingMain inserts many migration, assuming they're
+	// on the main line. This operation is necessary for compatibility before
+	// the `line` column was added to the migrations table.
+	MigrationInsertManyAssumingMain(ctx context.Context, versions []int) ([]*Migration, error)
 
 	NotifyMany(ctx context.Context, params *NotifyManyParams) error
 	PGAdvisoryXactLock(ctx context.Context, key int64) (*struct{}, error)
@@ -375,6 +395,11 @@ type Migration struct {
 	// API is not stable. DO NOT USE.
 	CreatedAt time.Time
 
+	// Line is the migration line that the migration belongs to.
+	//
+	// API is not stable. DO NOT USE.
+	Line string
+
 	// Version is the version of the migration.
 	//
 	// API is not stable. DO NOT USE.

riverdriver/riverdatabasesql/go.mod

@@ -11,7 +11,7 @@ replace github.com/riverqueue/river/rivertype => ../../rivertype
 require (
 	github.com/lib/pq v1.10.9
 	github.com/riverqueue/river/riverdriver v0.9.0
-	github.com/riverqueue/river/rivershared v0.0.0-20240707170519-d0685f5e0a5d
+	github.com/riverqueue/river/rivershared v0.0.0-20240707210043-f9063791ecb1
 	github.com/riverqueue/river/rivertype v0.9.0
 	github.com/stretchr/testify v1.9.0
 )