Tighten up schema of the migrations table a little more (#443)

I'd totally forgotten I'd even created it this way, but going back
through migrations recently I realized that the table's primary key is a
`bigserial`.

This isn't totally bad, and even has some advantages in that rows are
easier to target from a psql session, but in this instance it feels a
little on the sloppy side because it's fairly easy to conflate the
serial with migration `version` -- both are expected to be small
integers that a human could easily mix up.

Since we haven't shipped a new version yet containing the addition of
`line` in #435, here I propose that we modify that a little further to
reshape `river_migration` into a slightly smaller table with a compound
key on `(line, version)`:

    CREATE TABLE river_migration(
        created_at timestamptz NOT NULL DEFAULT NOW(),
        line TEXT NOT NULL,
        version bigint NOT NULL,
        CONSTRAINT line_length CHECK (char_length(line) > 0 AND char_length(line) < 128),
        CONSTRAINT version_gte_1 CHECK (version >= 1),
        PRIMARY KEY (line, version)
    );

This has two tiny side benefits that (1) the table is smaller (drops a
column), and (2) we don't need an extra index on `(line, version)` since
it's already covered by the primary key. Very similar in spirit to what
it was before, but I think a little cleaner in design.

internal/riverinternaltest/riverdrivertest/riverdrivertest.go

@@ -1972,7 +1972,6 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 
 		// 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, riverdriver.MigrationLineMain, migration1Fetched.Line)
 		require.Equal(t, migration1.Version, migration1Fetched.Version)
@@ -1999,7 +1998,6 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 
 		// 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)

riverdriver/river_driver_interface.go

@@ -385,11 +385,6 @@ type LeaderResignParams struct {
 //
 // API is not stable. DO NOT USE.
 type Migration struct {
-	// ID is an automatically generated primary key for the migration.
-	//
-	// API is not stable. DO NOT USE.
-	ID int
-
 	// CreatedAt is when the migration was initially created.
 	//
 	// API is not stable. DO NOT USE.

riverdriver/riverdatabasesql/migration/main/005_river_migration_add_line.down.sql

@@ -1,6 +1,36 @@
-DROP INDEX river_migration_line_version_idx;
-CREATE UNIQUE INDEX river_migration_version_idx ON river_migration USING btree(version);
+--
+-- If any non-main migration are present, 005 is considered irreversible.
+--
+
+DO
+$body$
+BEGIN
+    IF EXISTS (
+        SELECT *
+        FROM river_migration
+        WHERE line <> 'main'
+    ) THEN
+        RAISE EXCEPTION 'Found non-main migration lines in the database; version 005 migration is irreversible because it would result in loss of migration information.';
+    END IF;
+END;
+$body$
+LANGUAGE 'plpgsql'; 
 
 ALTER TABLE river_migration
-    DROP CONSTRAINT line_length,
-    DROP COLUMN line;
+    RENAME TO river_migration_old;
+
+CREATE TABLE river_migration(
+  id bigserial PRIMARY KEY,
+  created_at timestamptz NOT NULL DEFAULT NOW(),
+  version bigint NOT NULL,
+  CONSTRAINT version CHECK (version >= 1)
+);
+
+CREATE UNIQUE INDEX ON river_migration USING btree(version);
+
+INSERT INTO river_migration
+    (created_at, version)
+SELECT created_at, version
+FROM river_migration_old;
+
+DROP TABLE river_migration_old;

riverdriver/riverdatabasesql/migration/main/005_river_migration_add_line.up.sql

@@ -1,12 +1,18 @@
 ALTER TABLE river_migration
-    ADD COLUMN line text;
+    RENAME TO river_migration_old;
 
-UPDATE river_migration
-SET line = 'main';
+CREATE TABLE river_migration(
+    line TEXT NOT NULL,
+    version bigint NOT NULL,
+    created_at timestamptz NOT NULL DEFAULT NOW(),
+    CONSTRAINT line_length CHECK (char_length(line) > 0 AND char_length(line) < 128),
+    CONSTRAINT version_gte_1 CHECK (version >= 1),
+    PRIMARY KEY (line, version)
+);
 
-ALTER TABLE river_migration
-    ALTER COLUMN line SET NOT NULL,
-    ADD CONSTRAINT line_length CHECK (char_length(line) > 0 AND char_length(line) < 128);
+INSERT INTO river_migration
+    (created_at, line, version)
+SELECT created_at, 'main', version
+FROM river_migration_old;
 
-CREATE UNIQUE INDEX river_migration_line_version_idx ON river_migration USING btree(line, version);
-DROP INDEX river_migration_version_idx;
+DROP TABLE river_migration_old;

riverdriver/riverdatabasesql/river_database_sql.go

@@ -499,7 +499,6 @@ func (e *Executor) MigrationDeleteAssumingMainMany(ctx context.Context, versions
 	}
 	return sliceutil.Map(migrations, func(internal *dbsqlc.RiverMigrationDeleteAssumingMainManyRow) *riverdriver.Migration {
 		return &riverdriver.Migration{
-			ID:        int(internal.ID),
 			CreatedAt: internal.CreatedAt.UTC(),
 			Line:      riverdriver.MigrationLineMain,
 			Version:   int(internal.Version),
@@ -525,7 +524,6 @@ func (e *Executor) MigrationGetAllAssumingMain(ctx context.Context) ([]*riverdri
 	}
 	return sliceutil.Map(migrations, func(internal *dbsqlc.RiverMigrationGetAllAssumingMainRow) *riverdriver.Migration {
 		return &riverdriver.Migration{
-			ID:        int(internal.ID),
 			CreatedAt: internal.CreatedAt.UTC(),
 			Line:      riverdriver.MigrationLineMain,
 			Version:   int(internal.Version),
@@ -561,7 +559,6 @@ func (e *Executor) MigrationInsertManyAssumingMain(ctx context.Context, versions
 	}
 	return sliceutil.Map(migrations, func(internal *dbsqlc.RiverMigrationInsertManyAssumingMainRow) *riverdriver.Migration {
 		return &riverdriver.Migration{
-			ID:        int(internal.ID),
 			CreatedAt: internal.CreatedAt.UTC(),
 			Line:      riverdriver.MigrationLineMain,
 			Version:   int(internal.Version),
@@ -839,7 +836,6 @@ func mapSliceError[T any, R any](collection []T, mapFunc func(T) (R, error)) ([]
 
 func migrationFromInternal(internal *dbsqlc.RiverMigration) *riverdriver.Migration {
 	return &riverdriver.Migration{
-		ID:        int(internal.ID),
 		CreatedAt: internal.CreatedAt.UTC(),
 		Line:      internal.Line,
 		Version:   int(internal.Version),

riverdriver/riverpgxv5/internal/dbsqlc/river_migration.sql

@@ -1,17 +1,16 @@
 CREATE TABLE river_migration(
-    id bigserial PRIMARY KEY,
-    created_at timestamptz NOT NULL DEFAULT NOW(),
     line TEXT NOT NULL,
     version bigint NOT NULL,
+    created_at timestamptz NOT NULL DEFAULT NOW(),
     CONSTRAINT line_length CHECK (char_length(line) > 0 AND char_length(line) < 128),
-    CONSTRAINT version CHECK (version >= 1)
+    CONSTRAINT version_gte_1 CHECK (version >= 1),
+    PRIMARY KEY (line, version)
 );
 
 -- name: RiverMigrationDeleteAssumingMainMany :many
 DELETE FROM river_migration
 WHERE version = any(@version::bigint[])
 RETURNING
-    id,
     created_at,
     version;
 
@@ -29,7 +28,6 @@ RETURNING *;
 --
 -- name: RiverMigrationGetAllAssumingMain :many
 SELECT
-    id,
     created_at,
     version
 FROM river_migration
@@ -67,7 +65,6 @@ INSERT INTO river_migration (
 SELECT
     unnest(@version::bigint[])
 RETURNING
-    id,
     created_at,
     version;