Update database migrations, remove goose (#2264)

* Add new db migration

* Update migrations
Remove goose

* Add possibility to test direct upgrades

* Try to fix WASM test

* Add checks for specific migrations

* Remove AddMigration
Use WithTransaction
Add Dendrite version to table

* Fix linter issues

* Update tests

* Update comments, outdent if

* Namespace migrations

* Add direct upgrade tests, skipping over one version

* Split migrations

* Update go version in CI

* Fix copy&paste mistake

* Use contexts in migrations

Co-authored-by: kegsay <kegan@matrix.org>
Co-authored-by: Neil Alexander <neilalexander@users.noreply.github.com>

8 files changed, 73 insertions, 56 deletions

diff --git a/syncapi/storage/postgres/current_room_state_table.go b/syncapi/storage/postgres/current_room_state_table.go
index aae2d8c3..d13b7be4 100644
--- a/syncapi/storage/postgres/current_room_state_table.go
+++ b/syncapi/storage/postgres/current_room_state_table.go
@@ -23,6 +23,7 @@ import (
 	"github.com/lib/pq"
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/syncapi/storage/postgres/deltas"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
 	"github.com/matrix-org/dendrite/syncapi/types"
 	"github.com/matrix-org/gomatrixserverlib"
@@ -133,6 +134,17 @@ func NewPostgresCurrentRoomStateTable(db *sql.DB) (tables.CurrentRoomState, erro
 	if err != nil {
 		return nil, err
 	}
+
+	m := sqlutil.NewMigrator(db)
+	m.AddMigrations(sqlutil.Migration{
+		Version: "syncapi: add history visibility column (current_room_state)",
+		Up:      deltas.UpAddHistoryVisibilityColumnCurrentRoomState,
+	})
+	err = m.Up(context.Background())
+	if err != nil {
+		return nil, err
+	}
+
 	if s.upsertRoomStateStmt, err = db.Prepare(upsertRoomStateSQL); err != nil {
 		return nil, err
 	}
diff --git a/syncapi/storage/postgres/deltas/20201211125500_sequences.go b/syncapi/storage/postgres/deltas/20201211125500_sequences.go
index 7db524da..6303c947 100644
--- a/syncapi/storage/postgres/deltas/20201211125500_sequences.go
+++ b/syncapi/storage/postgres/deltas/20201211125500_sequences.go
@@ -15,24 +15,13 @@
 package deltas
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
-
-	"github.com/matrix-org/dendrite/internal/sqlutil"
-	"github.com/pressly/goose"
 )
 
-func LoadFromGoose() {
-	goose.AddMigration(UpFixSequences, DownFixSequences)
-	goose.AddMigration(UpRemoveSendToDeviceSentColumn, DownRemoveSendToDeviceSentColumn)
-}
-
-func LoadFixSequences(m *sqlutil.Migrations) {
-	m.AddMigration(UpFixSequences, DownFixSequences)
-}
-
-func UpFixSequences(tx *sql.Tx) error {
-	_, err := tx.Exec(`
+func UpFixSequences(ctx context.Context, tx *sql.Tx) error {
+	_, err := tx.ExecContext(ctx, `
 		-- We need to delete all of the existing receipts because the indexes
 		-- will be wrong, and we'll get primary key violations if we try to
 		-- reuse existing stream IDs from a different sequence.
@@ -49,8 +38,8 @@ func UpFixSequences(tx *sql.Tx) error {
 	return nil
 }
 
-func DownFixSequences(tx *sql.Tx) error {
-	_, err := tx.Exec(`
+func DownFixSequences(ctx context.Context, tx *sql.Tx) error {
+	_, err := tx.ExecContext(ctx, `
 		-- We need to delete all of the existing receipts because the indexes
 		-- will be wrong, and we'll get primary key violations if we try to
 		-- reuse existing stream IDs from a different sequence.
diff --git a/syncapi/storage/postgres/deltas/20210112130000_sendtodevice_sentcolumn.go b/syncapi/storage/postgres/deltas/20210112130000_sendtodevice_sentcolumn.go
index 3690eca8..77b083ae 100644
--- a/syncapi/storage/postgres/deltas/20210112130000_sendtodevice_sentcolumn.go
+++ b/syncapi/storage/postgres/deltas/20210112130000_sendtodevice_sentcolumn.go
@@ -15,18 +15,13 @@
 package deltas
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
-
-	"github.com/matrix-org/dendrite/internal/sqlutil"
 )
 
-func LoadRemoveSendToDeviceSentColumn(m *sqlutil.Migrations) {
-	m.AddMigration(UpRemoveSendToDeviceSentColumn, DownRemoveSendToDeviceSentColumn)
-}
-
-func UpRemoveSendToDeviceSentColumn(tx *sql.Tx) error {
-	_, err := tx.Exec(`
+func UpRemoveSendToDeviceSentColumn(ctx context.Context, tx *sql.Tx) error {
+	_, err := tx.ExecContext(ctx, `
 		ALTER TABLE syncapi_send_to_device
 		  DROP COLUMN IF EXISTS sent_by_token;
 	`)
@@ -36,8 +31,8 @@ func UpRemoveSendToDeviceSentColumn(tx *sql.Tx) error {
 	return nil
 }
 
-func DownRemoveSendToDeviceSentColumn(tx *sql.Tx) error {
-	_, err := tx.Exec(`
+func DownRemoveSendToDeviceSentColumn(ctx context.Context, tx *sql.Tx) error {
+	_, err := tx.ExecContext(ctx, `
 		ALTER TABLE syncapi_send_to_device
 		  ADD COLUMN IF NOT EXISTS sent_by_token TEXT;
 	`)
diff --git a/syncapi/storage/postgres/deltas/2022061412000000_history_visibility_column.go b/syncapi/storage/postgres/deltas/2022061412000000_history_visibility_column.go
index 2b4b6528..29008ade 100644
--- a/syncapi/storage/postgres/deltas/2022061412000000_history_visibility_column.go
+++ b/syncapi/storage/postgres/deltas/2022061412000000_history_visibility_column.go
@@ -15,20 +15,24 @@
 package deltas
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
-
-	"github.com/matrix-org/dendrite/internal/sqlutil"
 )
 
-func LoadAddHistoryVisibilityColumn(m *sqlutil.Migrations) {
-	m.AddMigration(UpAddHistoryVisibilityColumn, DownAddHistoryVisibilityColumn)
-}
-
-func UpAddHistoryVisibilityColumn(tx *sql.Tx) error {
-	_, err := tx.Exec(`
+func UpAddHistoryVisibilityColumnOutputRoomEvents(ctx context.Context, tx *sql.Tx) error {
+	_, err := tx.ExecContext(ctx, `
 		ALTER TABLE syncapi_output_room_events ADD COLUMN IF NOT EXISTS history_visibility SMALLINT NOT NULL DEFAULT 2;
 		UPDATE syncapi_output_room_events SET history_visibility = 4 WHERE type IN ('m.room.message', 'm.room.encrypted');
+	`)
+	if err != nil {
+		return fmt.Errorf("failed to execute upgrade: %w", err)
+	}
+	return nil
+}
+
+func UpAddHistoryVisibilityColumnCurrentRoomState(ctx context.Context, tx *sql.Tx) error {
+	_, err := tx.ExecContext(ctx, `
 		ALTER TABLE syncapi_current_room_state ADD COLUMN IF NOT EXISTS history_visibility SMALLINT NOT NULL DEFAULT 2;
 		UPDATE syncapi_current_room_state SET history_visibility = 4 WHERE type IN ('m.room.message', 'm.room.encrypted');
 	`)
@@ -38,8 +42,8 @@ func UpAddHistoryVisibilityColumn(tx *sql.Tx) error {
 	return nil
 }
 
-func DownAddHistoryVisibilityColumn(tx *sql.Tx) error {
-	_, err := tx.Exec(`
+func DownAddHistoryVisibilityColumn(ctx context.Context, tx *sql.Tx) error {
+	_, err := tx.ExecContext(ctx, `
 		ALTER TABLE syncapi_output_room_events DROP COLUMN IF EXISTS history_visibility;
 		ALTER TABLE syncapi_current_room_state DROP COLUMN IF EXISTS history_visibility;
 	`)
diff --git a/syncapi/storage/postgres/output_room_events_table.go b/syncapi/storage/postgres/output_room_events_table.go
index ddef2738..34ff6700 100644
--- a/syncapi/storage/postgres/output_room_events_table.go
+++ b/syncapi/storage/postgres/output_room_events_table.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/syncapi/storage/postgres/deltas"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
 	"github.com/matrix-org/dendrite/syncapi/types"
 
@@ -188,6 +189,17 @@ func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) {
 	if err != nil {
 		return nil, err
 	}
+
+	m := sqlutil.NewMigrator(db)
+	m.AddMigrations(sqlutil.Migration{
+		Version: "syncapi: add history visibility column (output_room_events)",
+		Up:      deltas.UpAddHistoryVisibilityColumnOutputRoomEvents,
+	})
+	err = m.Up(context.Background())
+	if err != nil {
+		return nil, err
+	}
+
 	return s, sqlutil.StatementList{
 		{&s.insertEventStmt, insertEventSQL},
 		{&s.selectEventsStmt, selectEventsSQL},
diff --git a/syncapi/storage/postgres/receipt_table.go b/syncapi/storage/postgres/receipt_table.go
index 2a42ffd7..bbddaa93 100644
--- a/syncapi/storage/postgres/receipt_table.go
+++ b/syncapi/storage/postgres/receipt_table.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/syncapi/storage/postgres/deltas"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
 	"github.com/matrix-org/dendrite/syncapi/types"
 	"github.com/matrix-org/gomatrixserverlib"
@@ -73,6 +74,15 @@ func NewPostgresReceiptsTable(db *sql.DB) (tables.Receipts, error) {
 	if err != nil {
 		return nil, err
 	}
+	m := sqlutil.NewMigrator(db)
+	m.AddMigrations(sqlutil.Migration{
+		Version: "syncapi: fix sequences",
+		Up:      deltas.UpFixSequences,
+	})
+	err = m.Up(context.Background())
+	if err != nil {
+		return nil, err
+	}
 	r := &receiptStatements{
 		db: db,
 	}
diff --git a/syncapi/storage/postgres/send_to_device_table.go b/syncapi/storage/postgres/send_to_device_table.go
index 96d6844f..2734fef3 100644
--- a/syncapi/storage/postgres/send_to_device_table.go
+++ b/syncapi/storage/postgres/send_to_device_table.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/syncapi/storage/postgres/deltas"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
 	"github.com/matrix-org/dendrite/syncapi/types"
 	"github.com/sirupsen/logrus"
@@ -76,6 +77,15 @@ func NewPostgresSendToDeviceTable(db *sql.DB) (tables.SendToDevice, error) {
 	if err != nil {
 		return nil, err
 	}
+	m := sqlutil.NewMigrator(db)
+	m.AddMigrations(sqlutil.Migration{
+		Version: "syncapi: drop sent_by_token",
+		Up:      deltas.UpRemoveSendToDeviceSentColumn,
+	})
+	err = m.Up(context.Background())
+	if err != nil {
+		return nil, err
+	}
 	if s.insertSendToDeviceMessageStmt, err = db.Prepare(insertSendToDeviceMessageSQL); err != nil {
 		return nil, err
 	}
diff --git a/syncapi/storage/postgres/syncserver.go b/syncapi/storage/postgres/syncserver.go
index c6121e41..a044716c 100644
--- a/syncapi/storage/postgres/syncserver.go
+++ b/syncapi/storage/postgres/syncserver.go
@@ -23,7 +23,6 @@ import (
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/setup/base"
 	"github.com/matrix-org/dendrite/setup/config"
-	"github.com/matrix-org/dendrite/syncapi/storage/postgres/deltas"
 	"github.com/matrix-org/dendrite/syncapi/storage/shared"
 )
 
@@ -42,13 +41,15 @@ func NewDatabase(base *base.BaseDendrite, dbProperties *config.DatabaseOptions)
 	if d.db, d.writer, err = base.DatabaseConnection(dbProperties, sqlutil.NewDummyWriter()); err != nil {
 		return nil, err
 	}
-	if _, err = d.db.Exec(outputRoomEventsSchema); err != nil {
+	accountData, err := NewPostgresAccountDataTable(d.db)
+	if err != nil {
 		return nil, err
 	}
-	if _, err = d.db.Exec(currentRoomStateSchema); err != nil {
+	events, err := NewPostgresEventsTable(d.db)
+	if err != nil {
 		return nil, err
 	}
-	accountData, err := NewPostgresAccountDataTable(d.db)
+	currState, err := NewPostgresCurrentRoomStateTable(d.db)
 	if err != nil {
 		return nil, err
 	}
@@ -96,22 +97,6 @@ func NewDatabase(base *base.BaseDendrite, dbProperties *config.DatabaseOptions)
 	if err != nil {
 		return nil, err
 	}
-	m := sqlutil.NewMigrations()
-	deltas.LoadFixSequences(m)
-	deltas.LoadRemoveSendToDeviceSentColumn(m)
-	deltas.LoadAddHistoryVisibilityColumn(m)
-	if err = m.RunDeltas(d.db, dbProperties); err != nil {
-		return nil, err
-	}
-	// prepare statements after the migrations have run
-	events, err := NewPostgresEventsTable(d.db)
-	if err != nil {
-		return nil, err
-	}
-	currState, err := NewPostgresCurrentRoomStateTable(d.db)
-	if err != nil {
-		return nil, err
-	}
 	d.Database = shared.Database{
 		DB:                  d.db,
 		Writer:              d.writer,