From a25b2ee39d3fcf959ba3f5180c944198071d6e82 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Tue, 14 Nov 2023 13:56:06 +0000 Subject: [PATCH] fix avatar updates; add e2e test for rejoining --- state/storage.go | 8 +- tests-e2e/client_test.go | 6 +- tests-e2e/state_updates_test.go | 238 ++++++++++++++++++++++++++++++++ 3 files changed, 248 insertions(+), 4 deletions(-) create mode 100644 tests-e2e/state_updates_test.go diff --git a/state/storage.go b/state/storage.go index 0f24379..b2cbcc8 100644 --- a/state/storage.go +++ b/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. diff --git a/tests-e2e/client_test.go b/tests-e2e/client_test.go index c8724a0..9ca4839 100644 --- a/tests-e2e/client_test.go +++ b/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)) }) } diff --git a/tests-e2e/state_updates_test.go b/tests-e2e/state_updates_test.go new file mode 100644 index 0000000..d06ca41 --- /dev/null +++ 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, + }, + }) +}