fix avatar updates; add e2e test for rejoining

state/storage.go

@@ -316,7 +316,7 @@ func (s *Storage) ResetMetadataState(metadata *internal.RoomMetadata) error {
 	FROM syncv3_events JOIN snapshot ON (
 		event_nid = ANY (ARRAY_CAT(events, membership_events))
 	)
-	WHERE (event_type IN ('m.room.name', 'm.room.avatar', 'm.room.canonical_alias') AND state_key = '')
+	WHERE (event_type IN ('m.room.name', 'm.room.avatar', 'm.room.canonical_alias', 'm.room.encryption') AND state_key = '')
 	   OR (event_type = 'm.room.member' AND membership IN ('join', '_join', 'invite', '_invite'))
 	ORDER BY event_nid ASC
 	;`, metadata.RoomID)
@@ -334,9 +334,11 @@ func (s *Storage) ResetMetadataState(metadata *internal.RoomMetadata) error {
 		case "m.room.name":
 			metadata.NameEvent = gjson.GetBytes(ev.JSON, "content.name").Str
 		case "m.room.avatar":
-			metadata.AvatarEvent = gjson.GetBytes(ev.JSON, "content.avatar_url").Str
+			metadata.AvatarEvent = gjson.GetBytes(ev.JSON, "content.url").Str
 		case "m.room.canonical_alias":
 			metadata.CanonicalAlias = gjson.GetBytes(ev.JSON, "content.alias").Str
+		case "m.room.encryption":
+			metadata.Encrypted = true
 		case "m.room.member":
 			heroMemberships.append(&events[i])
 			switch ev.Membership {
@@ -365,7 +367,7 @@ func (s *Storage) ResetMetadataState(metadata *internal.RoomMetadata) error {
 		metadata.Heroes = append(metadata.Heroes, hero)
 	}
 
-	// For now, don't bother reloading Encrypted, PredecessorID and UpgradedRoomID.
+	// For now, don't bother reloading PredecessorID and UpgradedRoomID.
 	// These shouldn't be changing during a room's lifetime in normal operation.
 
 	// We haven't updated LatestEventsByType because that's not part of the timeline.

tests-e2e/client_test.go

@@ -62,6 +62,9 @@ func (c *CSAPI) Scrollback(t *testing.T, roomID, prevBatch string, limit int) gj
 func (c *CSAPI) SlidingSync(t *testing.T, data sync3.Request, opts ...client.RequestOpt) (resBody *sync3.Response) {
 	t.Helper()
 	res := c.DoSlidingSync(t, data, opts...)
+	if res.StatusCode != 200 {
+		t.Fatalf("SlidingSync returned %v", res.Status)
+	}
 	body := client.ParseJSON(t, res)
 	if err := json.Unmarshal(body, &resBody); err != nil {
 		t.Fatalf("failed to unmarshal response: %v", err)
@@ -194,7 +197,8 @@ func (c *CSAPI) SlidingSyncUntilEvent(t *testing.T, pos string, data sync3.Reque
 				return nil
 			}
 		}
-		return fmt.Errorf("found room %s but missing event", roomID)
+		b, _ := json.Marshal(room.Timeline)
+		return fmt.Errorf("found room %s but missing event, timeline=%v", roomID, string(b))
 	})
 }
 

tests-e2e/state_updates_test.go

@@ -0,0 +1,238 @@
+package syncv3_test
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/matrix-org/complement/b"
+	"github.com/matrix-org/complement/client"
+	"github.com/matrix-org/sliding-sync/sync3"
+	"github.com/matrix-org/sliding-sync/testutils/m"
+)
+
+// The purpose of this test is to check that if Alice is in the room and is using the proxy, and then
+// she leaves the room so no _proxy user_ is in the room, and events are sent whilst she is gone, that
+// upon rejoining the proxy has A) the correct latest state and B) the correct latest timeline, such that
+// the prev_batch token is correct to get earlier messages.
+func TestRejoining(t *testing.T) {
+	alice := registerNamedUser(t, "alice")
+	bob := registerNamedUser(t, "bob")
+	charlie := registerNamedUser(t, "charlie")
+	doris := registerNamedUser(t, "doris")
+
+	// public chat => shared history visibility
+	roomID := bob.MustCreateRoom(t, map[string]interface{}{"preset": "public_chat"})
+	alice.JoinRoom(t, roomID, nil)
+	sendMessage(t, alice, roomID, "first message")
+	firstTopicEventID := sendTopic(t, bob, roomID, "First Topic")
+
+	// alice is the proxy user
+	res := alice.SlidingSync(t, sync3.Request{
+		Lists: map[string]sync3.RequestList{
+			"a": {
+				RoomSubscription: sync3.RoomSubscription{
+					TimelineLimit: 5,
+					RequiredState: [][2]string{{"*", "*"}}, // all state
+				},
+				Ranges:         sync3.SliceRanges{{0, 20}},
+				BumpEventTypes: []string{"m.room.message"},
+			},
+		},
+	})
+	m.MatchResponse(t, res, m.MatchRoomSubscriptionsStrict(
+		map[string][]m.RoomMatcher{
+			roomID: {
+				m.MatchJoinCount(2),
+				m.MatchRoomName(bob.Localpart),
+				// despite asking for 5 timeline events, we only get 1 on the initial sync so prev_batch is correct
+				MatchRoomTimeline([]Event{
+					{
+						ID: firstTopicEventID,
+					},
+				}),
+			},
+		},
+	))
+
+	// alice leaves the room => no proxy user in the room anymore!
+	alice.MustLeaveRoom(t, roomID)
+	res = alice.SlidingSyncUntilEvent(t, res.Pos, sync3.Request{}, roomID, Event{
+		Content: map[string]interface{}{
+			"membership": "leave",
+		},
+		Type:     "m.room.member",
+		StateKey: &alice.UserID,
+		Sender:   alice.UserID,
+	})
+
+	// now a few things happen:
+	// - charlie joins the room (ensures joins are reflected)
+	// - bob leaves the room (ensures leaves are reflected)
+	// - doris is invited to the room (ensures invites are reflected)
+	// - the topic is changed (ensures state updates)
+	// - the room avatar is set (ensure avatar field updates and new state is added)
+	// - 50 messages are sent (forcing a gappy sync)
+	charlie.MustJoinRoom(t, roomID, []string{"hs1"})
+	bob.MustInviteRoom(t, roomID, doris.UserID)
+	newTopic := "New Topic Name"
+	newAvatar := "mxc://example.org/JWEIFJgwEIhweiWJE"
+	sendTopic(t, bob, roomID, newTopic)
+	sendAvatar(t, bob, roomID, newAvatar)
+	bob.MustLeaveRoom(t, roomID)
+	messageEventIDs := make([]string, 0, 50)
+	for i := 0; i < 50; i++ {
+		messageEventIDs = append(messageEventIDs, sendMessage(t, charlie, roomID, fmt.Sprintf("message %d", i)))
+	}
+
+	// alice rejoins the room.
+	alice.MustJoinRoom(t, roomID, []string{"hs1"})
+
+	// we should get a 400 expired connection
+	start := time.Now()
+	for {
+		httpRes := alice.DoSlidingSync(t, sync3.Request{}, WithPos(res.Pos))
+		if httpRes.StatusCode == 400 {
+			t.Logf("connection expired")
+			break
+		}
+		if time.Since(start) > 5*time.Second {
+			t.Fatalf("did not get expired connection on rejoin")
+		}
+		body := client.ParseJSON(t, httpRes)
+		if err := json.Unmarshal(body, &res); err != nil {
+			t.Fatalf("failed to unmarshal response: %v", err)
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+
+	aliceJoin := Event{
+		Content: map[string]interface{}{
+			"displayname": alice.Localpart,
+			"membership":  "join",
+		},
+		Type:     "m.room.member",
+		StateKey: &alice.UserID,
+		Sender:   alice.UserID,
+	}
+	// ensure state is correct: new connection
+	res = alice.SlidingSyncUntilEvent(t, "", sync3.Request{
+		Lists: map[string]sync3.RequestList{
+			"a": {
+				RoomSubscription: sync3.RoomSubscription{
+					TimelineLimit: 5,
+					RequiredState: [][2]string{{"*", "*"}}, // all state
+				},
+				Ranges:         sync3.SliceRanges{{0, 20}},
+				BumpEventTypes: []string{"m.room.message"},
+			},
+		},
+	}, roomID, aliceJoin)
+	m.MatchResponse(t, res, m.MatchRoomSubscriptionsStrict(map[string][]m.RoomMatcher{
+		roomID: {
+			m.MatchInviteCount(1), // doris
+			m.MatchJoinCount(2),   // alice and charlie
+			MatchRoomTimeline([]Event{aliceJoin}),
+			m.MatchRoomAvatar(newAvatar),
+			MatchRoomRequiredState([]Event{
+				{
+					Type:     "m.room.topic",
+					StateKey: b.Ptr(""),
+					Content: map[string]interface{}{
+						"topic": newTopic,
+					},
+				},
+				{
+					Type:     "m.room.avatar",
+					StateKey: b.Ptr(""),
+					Content: map[string]interface{}{
+						"url": newAvatar,
+					},
+				},
+				aliceJoin,
+				{
+					Content: map[string]interface{}{
+						"displayname": charlie.Localpart,
+						"membership":  "join",
+					},
+					Type:     "m.room.member",
+					StateKey: &charlie.UserID,
+					Sender:   charlie.UserID,
+				},
+				{
+					Content: map[string]interface{}{
+						"membership":  "invite",
+						"displayname": doris.Localpart,
+					},
+					Type:     "m.room.member",
+					StateKey: &doris.UserID,
+					Sender:   bob.UserID,
+				},
+				{
+					Content: map[string]interface{}{
+						"membership": "leave",
+					},
+					Type:     "m.room.member",
+					StateKey: &bob.UserID,
+					Sender:   bob.UserID,
+				},
+			}),
+		},
+	}))
+
+	/*
+		// pull out the prev_batch and check we can backpaginate correctly
+		prevBatch := res.Rooms[roomID].PrevBatch
+		must.NotEqual(t, prevBatch, "", "missing prev_batch")
+		numScrollbackItems := 10
+		scrollback := alice.Scrollback(t, roomID, prevBatch, numScrollbackItems)
+		chunk := scrollback.Get("chunk").Array()
+		var sbEvents []json.RawMessage
+		for _, e := range chunk {
+			sbEvents = append(sbEvents, json.RawMessage(e.Raw))
+		}
+		must.Equal(t, len(chunk), 10, "chunk length mismatch")
+		var wantTimeline []Event
+		for i := 0; i < numScrollbackItems; i++ {
+			wantTimeline = append(wantTimeline, Event{
+				Type: "m.room.message",
+				Content: map[string]interface{}{
+					"msgtype": "m.text",
+					"body":    fmt.Sprintf("message %d", 40+i), // 40-49
+				},
+			})
+		}
+		must.NotError(t, "chunk mismatch", eventsEqual(wantTimeline, sbEvents)) */
+
+}
+
+func sendMessage(t *testing.T, client *CSAPI, roomID, text string) (eventID string) {
+	return client.Unsafe_SendEventUnsynced(t, roomID, b.Event{
+		Type: "m.room.message",
+		Content: map[string]interface{}{
+			"msgtype": "m.text",
+			"body":    text,
+		},
+	})
+}
+
+func sendTopic(t *testing.T, client *CSAPI, roomID, text string) (eventID string) {
+	return client.Unsafe_SendEventUnsynced(t, roomID, b.Event{
+		Type:     "m.room.topic",
+		StateKey: b.Ptr(""),
+		Content: map[string]interface{}{
+			"topic": text,
+		},
+	})
+}
+
+func sendAvatar(t *testing.T, client *CSAPI, roomID, mxcURI string) (eventID string) {
+	return client.Unsafe_SendEventUnsynced(t, roomID, b.Event{
+		Type:     "m.room.avatar",
+		StateKey: b.Ptr(""),
+		Content: map[string]interface{}{
+			"url": mxcURI,
+		},
+	})
+}