diff --git a/syncapi/storage/postgres/deltas/2022061412000000_history_visibility_column.go b/syncapi/storage/postgres/deltas/2022061412000000_history_visibility_column.go index 7582b1abe..d68ed8d5f 100644 --- a/syncapi/storage/postgres/deltas/2022061412000000_history_visibility_column.go +++ b/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) diff --git a/syncapi/storage/postgres/output_room_events_table.go b/syncapi/storage/postgres/output_room_events_table.go index 2a2c664f5..8f633640e 100644 --- a/syncapi/storage/postgres/output_room_events_table.go +++ b/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 { diff --git a/syncapi/storage/postgres/syncserver.go b/syncapi/storage/postgres/syncserver.go index a044716ce..979ff6647 100644 --- a/syncapi/storage/postgres/syncserver.go +++ b/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, diff --git a/syncapi/storage/sqlite3/deltas/2022061412000000_history_visibility_column.go b/syncapi/storage/sqlite3/deltas/2022061412000000_history_visibility_column.go index 3fa879db4..d23f07566 100644 --- a/syncapi/storage/sqlite3/deltas/2022061412000000_history_visibility_column.go +++ b/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) diff --git a/syncapi/storage/sqlite3/output_room_events_table.go b/syncapi/storage/sqlite3/output_room_events_table.go index 4f44d58b1..91fd35b5b 100644 --- a/syncapi/storage/sqlite3/output_room_events_table.go +++ b/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 { diff --git a/syncapi/storage/sqlite3/syncserver.go b/syncapi/storage/sqlite3/syncserver.go index 65b2bb38a..61210b84a 100644 --- a/syncapi/storage/sqlite3/syncserver.go +++ b/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, diff --git a/syncapi/storage/storage_test.go b/syncapi/storage/storage_test.go index df03a33c2..2a250ff77 100644 --- a/syncapi/storage/storage_test.go +++ b/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)