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>

16 files changed, 150 insertions, 120 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,
diff --git a/syncapi/storage/sqlite3/current_room_state_table.go b/syncapi/storage/sqlite3/current_room_state_table.go
index 501d0ee9..e19298ae 100644
--- a/syncapi/storage/sqlite3/current_room_state_table.go
+++ b/syncapi/storage/sqlite3/current_room_state_table.go
@@ -24,6 +24,7 @@ import (
 
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/syncapi/storage/sqlite3/deltas"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
 	"github.com/matrix-org/dendrite/syncapi/types"
 	"github.com/matrix-org/gomatrixserverlib"
@@ -120,6 +121,17 @@ func NewSqliteCurrentRoomStateTable(db *sql.DB, streamID *StreamIDStatements) (t
 	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/sqlite3/deltas/20201211125500_sequences.go b/syncapi/storage/sqlite3/deltas/20201211125500_sequences.go
index 8e7ebff8..f476335d 100644
--- a/syncapi/storage/sqlite3/deltas/20201211125500_sequences.go
+++ b/syncapi/storage/sqlite3/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.
@@ -45,8 +34,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/sqlite3/deltas/20210112130000_sendtodevice_sentcolumn.go b/syncapi/storage/sqlite3/deltas/20210112130000_sendtodevice_sentcolumn.go
index e0c51410..34cae224 100644
--- a/syncapi/storage/sqlite3/deltas/20210112130000_sendtodevice_sentcolumn.go
+++ b/syncapi/storage/sqlite3/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, `
 		CREATE TEMPORARY TABLE syncapi_send_to_device_backup(id, user_id, device_id, content);
 		INSERT INTO syncapi_send_to_device_backup SELECT id, user_id, device_id, content FROM syncapi_send_to_device;
 		DROP TABLE syncapi_send_to_device;
@@ -45,8 +40,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, `
 		CREATE TEMPORARY TABLE syncapi_send_to_device_backup(id, user_id, device_id, content);
 		INSERT INTO syncapi_send_to_device_backup SELECT id, user_id, device_id, content FROM syncapi_send_to_device;
 		DROP TABLE syncapi_send_to_device;
diff --git a/syncapi/storage/sqlite3/deltas/2022061412000000_history_visibility_column.go b/syncapi/storage/sqlite3/deltas/2022061412000000_history_visibility_column.go
index f6bcaddf..07917721 100644
--- a/syncapi/storage/sqlite3/deltas/2022061412000000_history_visibility_column.go
+++ b/syncapi/storage/sqlite3/deltas/2022061412000000_history_visibility_column.go
@@ -15,36 +15,36 @@
 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 {
+func UpAddHistoryVisibilityColumnOutputRoomEvents(ctx context.Context, tx *sql.Tx) error {
 	// SQLite doesn't have "if exists", so check if the column exists. If the query doesn't return an error, it already exists.
 	// Required for unit tests, as otherwise a duplicate column error will show up.
-	_, err := tx.Query("SELECT history_visibility FROM syncapi_output_room_events LIMIT 1")
+	_, err := tx.QueryContext(ctx, "SELECT history_visibility FROM syncapi_output_room_events LIMIT 1")
 	if err == nil {
 		return nil
 	}
-	_, err = tx.Exec(`
+	_, err = tx.ExecContext(ctx, `
 		ALTER TABLE syncapi_output_room_events ADD COLUMN 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
+}
 
-	_, err = tx.Query("SELECT history_visibility FROM syncapi_current_room_state LIMIT 1")
+func UpAddHistoryVisibilityColumnCurrentRoomState(ctx context.Context, tx *sql.Tx) error {
+	// SQLite doesn't have "if exists", so check if the column exists. If the query doesn't return an error, it already exists.
+	// Required for unit tests, as otherwise a duplicate column error will show up.
+	_, err := tx.QueryContext(ctx, "SELECT history_visibility FROM syncapi_current_room_state LIMIT 1")
 	if err == nil {
 		return nil
 	}
-	_, err = tx.Exec(`
+	_, err = tx.ExecContext(ctx, `
 		ALTER TABLE syncapi_current_room_state ADD COLUMN 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');
 	`)
@@ -54,25 +54,25 @@ func UpAddHistoryVisibilityColumn(tx *sql.Tx) error {
 	return nil
 }
 
-func DownAddHistoryVisibilityColumn(tx *sql.Tx) error {
+func DownAddHistoryVisibilityColumn(ctx context.Context, tx *sql.Tx) error {
 	// SQLite doesn't have "if exists", so check if the column exists.
-	_, err := tx.Query("SELECT history_visibility FROM syncapi_output_room_events LIMIT 1")
+	_, err := tx.QueryContext(ctx, "SELECT history_visibility FROM syncapi_output_room_events LIMIT 1")
 	if err != nil {
 		// The column probably doesn't exist
 		return nil
 	}
-	_, err = tx.Exec(`
+	_, err = tx.ExecContext(ctx, `
 		ALTER TABLE syncapi_output_room_events DROP COLUMN history_visibility;
 	`)
 	if err != nil {
 		return fmt.Errorf("failed to execute downgrade: %w", err)
 	}
-	_, err = tx.Query("SELECT history_visibility FROM syncapi_current_room_state LIMIT 1")
+	_, err = tx.QueryContext(ctx, "SELECT history_visibility FROM syncapi_current_room_state LIMIT 1")
 	if err != nil {
 		// The column probably doesn't exist
 		return nil
 	}
-	_, err = tx.Exec(`
+	_, err = tx.ExecContext(ctx, `
 		ALTER TABLE syncapi_current_room_state DROP COLUMN history_visibility;
 	`)
 	if err != nil {
diff --git a/syncapi/storage/sqlite3/output_room_events_table.go b/syncapi/storage/sqlite3/output_room_events_table.go
index b3dcb44c..de389fa9 100644
--- a/syncapi/storage/sqlite3/output_room_events_table.go
+++ b/syncapi/storage/sqlite3/output_room_events_table.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/syncapi/storage/sqlite3/deltas"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
 	"github.com/matrix-org/dendrite/syncapi/types"
 
@@ -136,6 +137,17 @@ func NewSqliteEventsTable(db *sql.DB, streamID *StreamIDStatements) (tables.Even
 	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.selectMaxEventIDStmt, selectMaxEventIDSQL},
diff --git a/syncapi/storage/sqlite3/receipt_table.go b/syncapi/storage/sqlite3/receipt_table.go
index bd778bf3..31adb005 100644
--- a/syncapi/storage/sqlite3/receipt_table.go
+++ b/syncapi/storage/sqlite3/receipt_table.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/syncapi/storage/sqlite3/deltas"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
 	"github.com/matrix-org/dendrite/syncapi/types"
 	"github.com/matrix-org/gomatrixserverlib"
@@ -70,6 +71,15 @@ func NewSqliteReceiptsTable(db *sql.DB, streamID *StreamIDStatements) (tables.Re
 	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,
 		streamIDStatements: streamID,
diff --git a/syncapi/storage/sqlite3/send_to_device_table.go b/syncapi/storage/sqlite3/send_to_device_table.go
index 5285acbe..d05d3fe7 100644
--- a/syncapi/storage/sqlite3/send_to_device_table.go
+++ b/syncapi/storage/sqlite3/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/sqlite3/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 NewSqliteSendToDeviceTable(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/sqlite3/syncserver.go b/syncapi/storage/sqlite3/syncserver.go
index 39ceec81..65b2bb38 100644
--- a/syncapi/storage/sqlite3/syncserver.go
+++ b/syncapi/storage/sqlite3/syncserver.go
@@ -22,7 +22,6 @@ import (
 	"github.com/matrix-org/dendrite/setup/base"
 	"github.com/matrix-org/dendrite/setup/config"
 	"github.com/matrix-org/dendrite/syncapi/storage/shared"
-	"github.com/matrix-org/dendrite/syncapi/storage/sqlite3/deltas"
 )
 
 // SyncServerDatasource represents a sync server datasource which manages
@@ -42,23 +41,25 @@ func NewDatabase(base *base.BaseDendrite, dbProperties *config.DatabaseOptions)
 	if d.db, d.writer, err = base.DatabaseConnection(dbProperties, sqlutil.NewExclusiveWriter()); err != nil {
 		return nil, err
 	}
-	if err = d.prepare(dbProperties); err != nil {
+	if err = d.prepare(); err != nil {
 		return nil, err
 	}
 	return &d, nil
 }
 
-func (d *SyncServerDatasource) prepare(dbProperties *config.DatabaseOptions) (err error) {
+func (d *SyncServerDatasource) prepare() (err error) {
 	if err = d.streamID.Prepare(d.db); err != nil {
 		return err
 	}
-	if _, err = d.db.Exec(outputRoomEventsSchema); err != nil {
+	accountData, err := NewSqliteAccountDataTable(d.db, &d.streamID)
+	if err != nil {
 		return err
 	}
-	if _, err = d.db.Exec(currentRoomStateSchema); err != nil {
+	events, err := NewSqliteEventsTable(d.db, &d.streamID)
+	if err != nil {
 		return err
 	}
-	accountData, err := NewSqliteAccountDataTable(d.db, &d.streamID)
+	roomState, err := NewSqliteCurrentRoomStateTable(d.db, &d.streamID)
 	if err != nil {
 		return err
 	}
@@ -106,22 +107,6 @@ func (d *SyncServerDatasource) prepare(dbProperties *config.DatabaseOptions) (er
 	if err != nil {
 		return err
 	}
-	m := sqlutil.NewMigrations()
-	deltas.LoadFixSequences(m)
-	deltas.LoadRemoveSendToDeviceSentColumn(m)
-	deltas.LoadAddHistoryVisibilityColumn(m)
-	if err = m.RunDeltas(d.db, dbProperties); err != nil {
-		return err
-	}
-	// prepare statements after the migrations have run
-	events, err := NewSqliteEventsTable(d.db, &d.streamID)
-	if err != nil {
-		return err
-	}
-	roomState, err := NewSqliteCurrentRoomStateTable(d.db, &d.streamID)
-	if err != nil {
-		return err
-	}
 	d.Database = shared.Database{
 		DB:                  d.db,
 		Writer:              d.writer,