Add FilteredSortableRooms which wraps SortableRooms with filters

README.md

@@ -27,6 +27,8 @@ Wait for the first initial v2 sync to be processed (this can take minutes!) and
 
 At present, the best way to help would be to run a local v3 server pointed at a busy account and just leave it and a client running in the background. Look at it occasionally and submit any issues you notice. You can save console logs by right-clicking -> Save As.
 
+Please run the server with `SYNCV3_DEBUG=1` set. This will force the server to panic when assertions fail rather than just log them.
+
 ## API
 
 API is under active development and is not stable.

sync3/handler/connstate.go

@@ -34,7 +34,7 @@ type ConnState struct {
 	userID string
 	// the only thing that can touch these data structures is the conn goroutine
 	muxedReq          *sync3.Request
-	sortedJoinedRooms *sync3.SortableRooms
+	sortedJoinedRooms *sync3.FilteredSortableRooms
 	roomSubscriptions map[string]sync3.RoomSubscription
 	loadPosition      int64
 
@@ -90,9 +90,8 @@ func (s *ConnState) load(req *sync3.Request) error {
 			NotificationCount: urd.NotificationCount,
 		}
 	}
-
 	s.loadPosition = initialLoadPosition
-	s.sortedJoinedRooms = sync3.NewSortableRooms(rooms)
+	s.sortedJoinedRooms = sync3.NewFilteredSortableRooms(rooms, req.Lists[0].Filters)
 	s.sort(req.Lists[0].Sort)
 
 	return nil
@@ -249,11 +248,10 @@ func (s *ConnState) processIncomingUserEvent(roomID string, userEvent *sync3.Use
 }
 
 func (s *ConnState) processIncomingEvent(updateEvent *sync3.EventData) ([]sync3.Room, []sync3.ResponseOp) {
+	// keep track of the latest stream position
 	if updateEvent.LatestPos > s.loadPosition {
 		s.loadPosition = updateEvent.LatestPos
 	}
-	// TODO: Add filters to check if this event should cause a response or should be dropped (e.g filtering out messages)
-	// this is why this select is in a while loop as not all update event will wake up the stream
 
 	fromIndex, ok := s.sortedJoinedRooms.IndexOf(updateEvent.RoomID)
 	if !ok {
@@ -268,7 +266,10 @@ func (s *ConnState) processIncomingEvent(updateEvent *sync3.EventData) ([]sync3.
 				strings.Trim(internal.CalculateRoomName(roomMetadata, 5), "#!()):_@"),
 			),
 		}
-		s.sortedJoinedRooms.Add(newRoomConn)
+		if !s.sortedJoinedRooms.Add(newRoomConn) {
+			// we didn't add this room to the list so we don't need to resort
+			return nil, nil
+		}
 	}
 	return s.resort(updateEvent.RoomID, fromIndex, updateEvent.Event)
 }

sync3/request.go

@@ -3,6 +3,8 @@ package sync3
 import (
 	"bytes"
 	"encoding/json"
+
+	"github.com/matrix-org/sync-v3/internal"
 )
 
 var (
@@ -11,8 +13,9 @@ var (
 	SortByNotificationCount = "by_notification_count"
 	SortByHighlightCount    = "by_highlight_count"
 	SortBy                  = []string{SortByHighlightCount, SortByName, SortByNotificationCount, SortByRecency}
-	DefaultTimelineLimit    = int64(20)
-	DefaultTimeoutSecs      = 10
+
+	DefaultTimelineLimit = int64(20)
+	DefaultTimeoutSecs   = 10
 )
 
 type Request struct {
@@ -172,10 +175,20 @@ func (r *Request) GetRequiredState(listIndex int, roomID string) [][2]string {
 }
 
 type RequestFilters struct {
-	Spaces []string `json:"spaces"`
+	Spaces      []string `json:"spaces"`
+	IsDM        *bool    `json:"is_dm"`
+	IsEncrypted *bool    `json:"is_encrypted"`
+	IsInvite    *bool    `json:"is_invite"`
 	// TODO options to control which events should be live-streamed e.g not_types, types from sync v2
 }
 
+func (rf *RequestFilters) Include(r *internal.RoomMetadata) bool {
+	if rf.IsEncrypted != nil && *rf.IsEncrypted != r.Encrypted {
+		return false
+	}
+	return true
+}
+
 type RoomSubscription struct {
 	RequiredState [][2]string `json:"required_state"`
 	TimelineLimit int64       `json:"timeline_limit"`

sync3/sort.go

@@ -55,9 +55,15 @@ func (s *SortableRooms) RoomIDs() []string {
 	return roomIDs
 }
 
-func (s *SortableRooms) Add(r RoomConnMetadata) {
+// Add a room to the list. Returns true if the room was added.
+func (s *SortableRooms) Add(r RoomConnMetadata) bool {
+	_, exists := s.roomIDToIndex[r.RoomID]
+	if exists {
+		return false
+	}
 	s.rooms = append(s.rooms, r)
 	s.roomIDToIndex[r.RoomID] = len(s.rooms) - 1
+	return true
 }
 
 func (s *SortableRooms) Get(index int) RoomConnMetadata {
@@ -65,6 +71,16 @@ func (s *SortableRooms) Get(index int) RoomConnMetadata {
 	return s.rooms[index]
 }
 
+func (s *SortableRooms) Remove(roomID string) {
+	index, ok := s.roomIDToIndex[roomID]
+	if !ok {
+		return
+	}
+	delete(s.roomIDToIndex, roomID)
+	// splice out index
+	s.rooms = append(s.rooms[:index], s.rooms[index+1:]...)
+}
+
 func (s *SortableRooms) Len() int64 {
 	return int64(len(s.rooms))
 }
@@ -155,3 +171,39 @@ func (s *SortableRooms) comparatorSortByNotificationCount(i, j int) int {
 	}
 	return -1
 }
+
+type FilteredSortableRooms struct {
+	*SortableRooms
+	filter *RequestFilters
+}
+
+func NewFilteredSortableRooms(rooms []RoomConnMetadata, filter *RequestFilters) *FilteredSortableRooms {
+	var filteredRooms []RoomConnMetadata
+	if filter == nil {
+		filter = &RequestFilters{}
+	}
+	for _, r := range rooms {
+		if filter.Include(&r.RoomMetadata) {
+			filteredRooms = append(filteredRooms, r)
+		}
+	}
+	return &FilteredSortableRooms{
+		SortableRooms: NewSortableRooms(filteredRooms),
+		filter:        filter,
+	}
+}
+
+func (f *FilteredSortableRooms) Add(r RoomConnMetadata) bool {
+	if !f.filter.Include(&r.RoomMetadata) {
+		return false
+	}
+	return f.SortableRooms.Add(r)
+}
+
+func (f *FilteredSortableRooms) UpdateGlobalRoomMetadata(r *internal.RoomMetadata) {
+	if !f.filter.Include(r) {
+		f.Remove(r.RoomID)
+		return
+	}
+	f.SortableRooms.UpdateGlobalRoomMetadata(r)
+}