Implement history visibility on /messages, /context, /sync (#2511)

* Add possibility to set history_visibility and user AccountType

* Add new DB queries

* Add actual history_visibility changes for /messages

* Add passing tests

* Extract check function

* Cleanup

* Cleanup

* Fix build on 386

* Move ApplyHistoryVisibilityFilter to internal

* Move queries to topology table

* Add filtering to /sync and /context
Some cleanup

* Add passing tests; Remove failing tests :(

* Re-add passing tests

* Move filtering to own function to avoid duplication

* Re-add passing test

* Use newly added GMSL HistoryVisibility

* Update gomatrixserverlib

* Set the visibility when creating events

* Default to shared history visibility

* Remove unused query

* Update history visibility checks to use gmsl
Update tests

* Remove unused statement

* Update migrations to set "correct" history visibility

* Add method to fetch the membership at a given event

* Tweaks and logging

* Use actual internal rsAPI, default to shared visibility in tests

* Revert "Move queries to topology table"

This reverts commit 4f0d41be9c194a46379796435ce73e79203edbd6.

* Remove noise/unneeded code

* More cleanup

* Try to optimize database requests

* Fix imports

* PR peview fixes/changes

* Move setting history visibility to own migration, be more restrictive

* Fix unit tests

* Lint

* Fix missing entries

* Tweaks for incremental syncs

* Adapt generic changes

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

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

@@ -17,7 +17,10 @@ package deltas
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 	"fmt"
+
+	"github.com/matrix-org/gomatrixserverlib"
 )
 
 func UpAddHistoryVisibilityColumnOutputRoomEvents(ctx context.Context, tx *sql.Tx) error {
@@ -37,6 +40,27 @@ func UpAddHistoryVisibilityColumnOutputRoomEvents(ctx context.Context, tx *sql.T
 	return nil
 }
 
+// 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)
+	if err != nil {
+		return err
+	}
+
+	// update the history visibility
+	for roomID, hisVis := range historyVisibilities {
+		_, err = tx.ExecContext(ctx, `UPDATE syncapi_output_room_events SET history_visibility = $1 
+                        WHERE type IN ('m.room.message', 'm.room.encrypted') AND room_id = $2 AND history_visibility <> $1`, hisVis, roomID)
+		if err != nil {
+			return fmt.Errorf("failed to update history visibility: %w", err)
+		}
+	}
+
+	return nil
+}
+
 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.
@@ -51,9 +75,40 @@ func UpAddHistoryVisibilityColumnCurrentRoomState(ctx context.Context, tx *sql.T
 	if err != nil {
 		return fmt.Errorf("failed to execute upgrade: %w", err)
 	}
+
 	return nil
 }
 
+// currentHistoryVisibilities returns a map from roomID to current history visibility.
+// If the history visibility was changed after room creation, defaults to joined.
+func currentHistoryVisibilities(ctx context.Context, tx *sql.Tx) (map[string]gomatrixserverlib.HistoryVisibility, error) {
+	rows, err := tx.QueryContext(ctx, `SELECT DISTINCT room_id, headered_event_json FROM syncapi_current_room_state
+		WHERE type = 'm.room.history_visibility' AND state_key = '';
+`)
+	if err != nil {
+		return nil, fmt.Errorf("failed to query current room state: %w", err)
+	}
+	defer rows.Close() // nolint: errcheck
+	var eventBytes []byte
+	var roomID string
+	var event gomatrixserverlib.HeaderedEvent
+	var hisVis gomatrixserverlib.HistoryVisibility
+	historyVisibilities := make(map[string]gomatrixserverlib.HistoryVisibility)
+	for rows.Next() {
+		if err = rows.Scan(&roomID, &eventBytes); err != nil {
+			return nil, fmt.Errorf("failed to scan row: %w", err)
+		}
+		if err = json.Unmarshal(eventBytes, &event); err != nil {
+			return nil, fmt.Errorf("failed to unmarshal event: %w", err)
+		}
+		historyVisibilities[roomID] = gomatrixserverlib.HistoryVisibilityJoined
+		if hisVis, err = event.HistoryVisibility(); err == nil && event.Depth() < 10 {
+			historyVisibilities[roomID] = hisVis
+		}
+	}
+	return historyVisibilities, nil
+}
+
 func DownAddHistoryVisibilityColumn(ctx context.Context, tx *sql.Tx) error {
 	// SQLite doesn't have "if exists", so check if the column exists.
 	_, err := tx.QueryContext(ctx, "SELECT history_visibility FROM syncapi_output_room_events LIMIT 1")

syncapi/storage/sqlite3/memberships_table.go

@@ -66,11 +66,15 @@ const selectMembershipCountSQL = "" +
 const selectHeroesSQL = "" +
 	"SELECT DISTINCT user_id FROM syncapi_memberships WHERE room_id = $1 AND user_id != $2 AND membership IN ($3) LIMIT 5"
 
+const selectMembershipBeforeSQL = "" +
+	"SELECT membership, topological_pos FROM syncapi_memberships WHERE room_id = $1 and user_id = $2 AND topological_pos <= $3 ORDER BY topological_pos DESC LIMIT 1"
+
 type membershipsStatements struct {
 	db                        *sql.DB
 	upsertMembershipStmt      *sql.Stmt
 	selectMembershipCountStmt *sql.Stmt
 	//selectHeroesStmt          *sql.Stmt - prepared at runtime due to variadic
+	selectMembershipForUserStmt *sql.Stmt
 }
 
 func NewSqliteMembershipsTable(db *sql.DB) (tables.Memberships, error) {
@@ -84,6 +88,7 @@ func NewSqliteMembershipsTable(db *sql.DB) (tables.Memberships, error) {
 	return s, sqlutil.StatementList{
 		{&s.upsertMembershipStmt, upsertMembershipSQL},
 		{&s.selectMembershipCountStmt, selectMembershipCountSQL},
+		{&s.selectMembershipForUserStmt, selectMembershipBeforeSQL},
 		// {&s.selectHeroesStmt, selectHeroesSQL}, - prepared at runtime due to variadic
 	}.Prepare(db)
 }
@@ -148,3 +153,20 @@ func (s *membershipsStatements) SelectHeroes(
 	}
 	return heroes, rows.Err()
 }
+
+// SelectMembershipForUser returns the membership of the user before and including the given position. If no membership can be found
+// returns "leave", the topological position and no error. If an error occurs, other than sql.ErrNoRows, returns that and an empty
+// string as the membership.
+func (s *membershipsStatements) SelectMembershipForUser(
+	ctx context.Context, txn *sql.Tx, roomID, userID string, pos int64,
+) (membership string, topologyPos int, err error) {
+	stmt := sqlutil.TxStmt(txn, s.selectMembershipForUserStmt)
+	err = stmt.QueryRowContext(ctx, roomID, userID, pos).Scan(&membership, &topologyPos)
+	if err != nil {
+		if err == sql.ErrNoRows {
+			return "leave", 0, nil
+		}
+		return "", 0, err
+	}
+	return membership, topologyPos, nil
+}

syncapi/storage/sqlite3/output_room_events_table.go

@@ -139,10 +139,12 @@ func NewSqliteEventsTable(db *sql.DB, streamID *StreamIDStatements) (tables.Even
 	}
 
 	m := sqlutil.NewMigrator(db)
-	m.AddMigrations(sqlutil.Migration{
-		Version: "syncapi: add history visibility column (output_room_events)",
-		Up:      deltas.UpAddHistoryVisibilityColumnOutputRoomEvents,
-	})
+	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

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,