Fix unit tests

syncapi/storage/postgres/deltas/2022061412000000_history_visibility_column.go

@@ -35,6 +35,7 @@ func UpAddHistoryVisibilityColumnOutputRoomEvents(ctx context.Context, tx *sql.T
 }
 
 // UpSetHistoryVisibility sets the history visibility for already stored events.
+// Requires current_room_state and output_room_events to be created.
 func UpSetHistoryVisibility(ctx context.Context, tx *sql.Tx) error {
 	// get the current room history visibilities
 	historyVisibilities, err := currentHistoryVisibilities(ctx, tx)

syncapi/storage/postgres/output_room_events_table.go

@@ -196,10 +196,6 @@ func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) {
 			Version: "syncapi: add history visibility column (output_room_events)",
 			Up:      deltas.UpAddHistoryVisibilityColumnOutputRoomEvents,
 		},
-		sqlutil.Migration{
-			Version: "syncapi: set history visibility for existing events",
-			Up:      deltas.UpSetHistoryVisibility,
-		},
 	)
 	err = m.Up(context.Background())
 	if err != nil {

syncapi/storage/postgres/syncserver.go

@@ -23,6 +23,7 @@ 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"
 )
 
@@ -97,6 +98,20 @@ func NewDatabase(base *base.BaseDendrite, dbProperties *config.DatabaseOptions)
 	if err != nil {
 		return nil, err
 	}
+
+	// apply migrations which need multiple tables
+	m := sqlutil.NewMigrator(d.db)
+	m.AddMigrations(
+		sqlutil.Migration{
+			Version: "syncapi: set history visibility for existing events",
+			Up:      deltas.UpSetHistoryVisibility, // Requires current_room_state and output_room_events to be created.
+		},
+	)
+	err = m.Up(base.Context())
+	if err != nil {
+		return nil, err
+	}
+
 	d.Database = shared.Database{
 		DB:                  d.db,
 		Writer:              d.writer,

syncapi/storage/sqlite3/deltas/2022061412000000_history_visibility_column.go

@@ -41,6 +41,7 @@ func UpAddHistoryVisibilityColumnOutputRoomEvents(ctx context.Context, tx *sql.T
 }
 
 // UpSetHistoryVisibility sets the history visibility for already stored events.
+// Requires current_room_state and output_room_events to be created.
 func UpSetHistoryVisibility(ctx context.Context, tx *sql.Tx) error {
 	// get the current room history visibilities
 	historyVisibilities, err := currentHistoryVisibilities(ctx, tx)

syncapi/storage/sqlite3/output_room_events_table.go

@@ -144,10 +144,6 @@ func NewSqliteEventsTable(db *sql.DB, streamID *StreamIDStatements) (tables.Even
 			Version: "syncapi: add history visibility column (output_room_events)",
 			Up:      deltas.UpAddHistoryVisibilityColumnOutputRoomEvents,
 		},
-		sqlutil.Migration{
-			Version: "syncapi: set history visibility for existing events",
-			Up:      deltas.UpSetHistoryVisibility,
-		},
 	)
 	err = m.Up(context.Background())
 	if err != nil {

syncapi/storage/sqlite3/syncserver.go

@@ -16,12 +16,14 @@
 package sqlite3
 
 import (
+	"context"
 	"database/sql"
 
 	"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/shared"
+	"github.com/matrix-org/dendrite/syncapi/storage/sqlite3/deltas"
 )
 
 // SyncServerDatasource represents a sync server datasource which manages
@@ -41,13 +43,13 @@ 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(); err != nil {
+	if err = d.prepare(base.Context()); err != nil {
 		return nil, err
 	}
 	return &d, nil
 }
 
-func (d *SyncServerDatasource) prepare() (err error) {
+func (d *SyncServerDatasource) prepare(ctx context.Context) (err error) {
 	if err = d.streamID.Prepare(d.db); err != nil {
 		return err
 	}
@@ -107,6 +109,19 @@ func (d *SyncServerDatasource) prepare() (err error) {
 	if err != nil {
 		return err
 	}
+
+	// apply migrations which need multiple tables
+	m := sqlutil.NewMigrator(d.db)
+	m.AddMigrations(
+		sqlutil.Migration{
+			Version: "syncapi: set history visibility for existing events",
+			Up:      deltas.UpSetHistoryVisibility, // Requires current_room_state and output_room_events to be created.
+		},
+	)
+	err = m.Up(ctx)
+	if err != nil {
+		return err
+	}
 	d.Database = shared.Database{
 		DB:                  d.db,
 		Writer:              d.writer,

syncapi/storage/storage_test.go

@@ -12,20 +12,22 @@ import (
 	"github.com/matrix-org/dendrite/syncapi/storage"
 	"github.com/matrix-org/dendrite/syncapi/types"
 	"github.com/matrix-org/dendrite/test"
+	"github.com/matrix-org/dendrite/test/testrig"
 	"github.com/matrix-org/gomatrixserverlib"
 )
 
 var ctx = context.Background()
 
-func MustCreateDatabase(t *testing.T, dbType test.DBType) (storage.Database, func()) {
+func MustCreateDatabase(t *testing.T, dbType test.DBType) (storage.Database, func(), func()) {
 	connStr, close := test.PrepareDBConnectionString(t, dbType)
-	db, err := storage.NewSyncServerDatasource(nil, &config.DatabaseOptions{
+	base, closeBase := testrig.CreateBaseDendrite(t, dbType)
+	db, err := storage.NewSyncServerDatasource(base, &config.DatabaseOptions{
 		ConnectionString: config.DataSource(connStr),
 	})
 	if err != nil {
 		t.Fatalf("NewSyncServerDatasource returned %s", err)
 	}
-	return db, close
+	return db, close, closeBase
 }
 
 func MustWriteEvents(t *testing.T, db storage.Database, events []*gomatrixserverlib.HeaderedEvent) (positions []types.StreamPosition) {
@@ -51,8 +53,9 @@ func TestWriteEvents(t *testing.T) {
 	test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
 		alice := test.NewUser(t)
 		r := test.NewRoom(t, alice)
-		db, close := MustCreateDatabase(t, dbType)
+		db, close, closeBase := MustCreateDatabase(t, dbType)
 		defer close()
+		defer closeBase()
 		MustWriteEvents(t, db, r.Events())
 	})
 }
@@ -60,8 +63,9 @@ func TestWriteEvents(t *testing.T) {
 // These tests assert basic functionality of RecentEvents for PDUs
 func TestRecentEventsPDU(t *testing.T) {
 	test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
-		db, close := MustCreateDatabase(t, dbType)
+		db, close, closeBase := MustCreateDatabase(t, dbType)
 		defer close()
+		defer closeBase()
 		alice := test.NewUser(t)
 		// dummy room to make sure SQL queries are filtering on room ID
 		MustWriteEvents(t, db, test.NewRoom(t, alice).Events())
@@ -163,8 +167,9 @@ func TestRecentEventsPDU(t *testing.T) {
 // The purpose of this test is to ensure that backfill does indeed go backwards, using a topology token
 func TestGetEventsInRangeWithTopologyToken(t *testing.T) {
 	test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
-		db, close := MustCreateDatabase(t, dbType)
+		db, close, closeBase := MustCreateDatabase(t, dbType)
 		defer close()
+		defer closeBase()
 		alice := test.NewUser(t)
 		r := test.NewRoom(t, alice)
 		for i := 0; i < 10; i++ {
@@ -404,8 +409,9 @@ func TestSendToDeviceBehaviour(t *testing.T) {
 	bob := test.NewUser(t)
 	deviceID := "one"
 	test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
-		db, close := MustCreateDatabase(t, dbType)
+		db, close, closeBase := MustCreateDatabase(t, dbType)
 		defer close()
+		defer closeBase()
 		// At this point there should be no messages. We haven't sent anything
 		// yet.
 		_, events, err := db.SendToDeviceUpdatesForSync(ctx, alice.ID, deviceID, 0, 100)