read receipts: ensure that read receipts show up via /sync (#437)

* read receipts: ensure that read receipts show up via /sync

Signed-off-by: Sumner Evans <sumner@beeper.com>

* typing: use new SyncEphemeralHas helper function

Signed-off-by: Sumner Evans <sumner@beeper.com>

Signed-off-by: Sumner Evans <sumner@beeper.com>

internal/client/client.go

@@ -682,6 +682,18 @@ func SyncTimelineHasEventID(roomID string, eventID string) SyncCheckOpt {
 	})
 }
 
+func SyncEphemeralHas(roomID string, check func(gjson.Result) bool) SyncCheckOpt {
+	return func(clientUserID string, topLevelSyncJSON gjson.Result) error {
+		err := loopArray(
+			topLevelSyncJSON, "rooms.join."+GjsonEscape(roomID)+".ephemeral.events", check,
+		)
+		if err == nil {
+			return nil
+		}
+		return fmt.Errorf("SyncEphemeralHas(%s): %s", roomID, err)
+	}
+}
+
 // Checks that `userID` gets invited to `roomID`.
 //
 // This checks different parts of the /sync response depending on the client making the request.
@@ -714,15 +726,27 @@ func SyncInvitedTo(userID, roomID string) SyncCheckOpt {
 
 // Check that `userID` gets joined to `roomID` by inspecting the join timeline for a membership event
 func SyncJoinedTo(userID, roomID string) SyncCheckOpt {
+	checkJoined := func(ev gjson.Result) bool {
+		return ev.Get("type").Str == "m.room.member" && ev.Get("state_key").Str == userID && ev.Get("content.membership").Str == "join"
+	}
 	return func(clientUserID string, topLevelSyncJSON gjson.Result) error {
-		// awkward wrapping to get the error message correct at the start :/
-		err := SyncTimelineHas(roomID, func(ev gjson.Result) bool {
-			return ev.Get("type").Str == "m.room.member" && ev.Get("state_key").Str == userID && ev.Get("content.membership").Str == "join"
-		})(clientUserID, topLevelSyncJSON)
+		// Check both the timeline and the state events for the join event
+		// since on initial sync, the state events may only be in
+		// <room>.state.events.
+		err := loopArray(
+			topLevelSyncJSON, "rooms.join."+GjsonEscape(roomID)+".timeline.events", checkJoined,
+		)
 		if err == nil {
 			return nil
 		}
-		return fmt.Errorf("SyncJoinedTo(%s,%s): %s", userID, roomID, err)
+
+		err = loopArray(
+			topLevelSyncJSON, "rooms.join."+GjsonEscape(roomID)+".state.events", checkJoined,
+		)
+		if err == nil {
+			return nil
+		}
+		return fmt.Errorf("SyncJoinedTo(%s): %s", roomID, err)
 	}
 }
 
@@ -757,6 +781,21 @@ func SyncGlobalAccountDataHas(check func(gjson.Result) bool) SyncCheckOpt {
 	}
 }
 
+// Calls the `check` function for each account data event for the given room,
+// and returns with success if the `check` function returns true for at least
+// one event.
+func SyncRoomAccountDataHas(roomID string, check func(gjson.Result) bool) SyncCheckOpt {
+	return func(clientUserID string, topLevelSyncJSON gjson.Result) error {
+		err := loopArray(
+			topLevelSyncJSON, "rooms.join."+GjsonEscape(roomID)+".account_data.events", check,
+		)
+		if err == nil {
+			return nil
+		}
+		return fmt.Errorf("SyncRoomAccountDataHas(%s): %s", roomID, err)
+	}
+}
+
 func loopArray(object gjson.Result, key string, check func(gjson.Result) bool) error {
 	array := object.Get(key)
 	if !array.Exists() {

tests/csapi/apidoc_room_receipts_test.go

@@ -5,56 +5,68 @@ import (
 
 	"github.com/matrix-org/complement/internal/b"
 	"github.com/matrix-org/complement/internal/client"
+	"github.com/matrix-org/complement/internal/docker"
 	"github.com/tidwall/gjson"
 )
 
 // tests/10apidoc/37room-receipts.pl
 
+func createRoomForReadReceipts(t *testing.T, c *client.CSAPI, deployment *docker.Deployment) (string, string) {
+	roomID := c.CreateRoom(t, map[string]interface{}{"preset": "public_chat"})
+
+	c.MustSyncUntil(t, client.SyncReq{}, client.SyncJoinedTo(c.UserID, roomID))
+
+	eventID := c.SendEventSynced(t, roomID, b.Event{
+		Type: "m.room.message",
+		Content: map[string]interface{}{
+			"msgtype": "m.text",
+			"body":    "Hello world!",
+		},
+	})
+
+	return roomID, eventID
+}
+
+func syncHasReadReceipt(roomID, userID, eventID string) client.SyncCheckOpt {
+	return client.SyncEphemeralHas(roomID, func(result gjson.Result) bool {
+		return result.Get("type").Str == "m.receipt" &&
+			result.Get("content").Get(eventID).Get(`m\.read`).Get(userID).Exists()
+	})
+}
+
 // sytest: POST /rooms/:room_id/receipt can create receipts
 func TestRoomReceipts(t *testing.T) {
 	deployment := Deploy(t, b.BlueprintAlice)
 	defer deployment.Destroy(t)
-
 	alice := deployment.Client(t, "hs1", "@alice:hs1")
-	roomID := alice.CreateRoom(t, map[string]interface{}{"preset": "public_chat"})
-
-	eventID := ""
-	alice.MustSyncUntil(t, client.SyncReq{}, client.SyncTimelineHas(roomID, func(result gjson.Result) bool {
-		if result.Get("type").Str == "m.room.member" {
-			eventID = result.Get("event_id").Str
-			return true
-		}
-		return false
-	}))
-	if eventID == "" {
-		t.Fatal("did not find an event_id")
-	}
+	roomID, eventID := createRoomForReadReceipts(t, alice, deployment)
+
 	alice.MustDoFunc(t, "POST", []string{"_matrix", "client", "v3", "rooms", roomID, "receipt", "m.read", eventID}, client.WithJSONBody(t, struct{}{}))
+
+	// Make sure the read receipt shows up in sync.
+	alice.MustSyncUntil(t, client.SyncReq{}, syncHasReadReceipt(roomID, alice.UserID, eventID))
 }
 
 // sytest: POST /rooms/:room_id/read_markers can create read marker
 func TestRoomReadMarkers(t *testing.T) {
 	deployment := Deploy(t, b.BlueprintAlice)
 	defer deployment.Destroy(t)
-
 	alice := deployment.Client(t, "hs1", "@alice:hs1")
-	roomID := alice.CreateRoom(t, map[string]interface{}{"preset": "public_chat"})
-
-	eventID := ""
-	alice.MustSyncUntil(t, client.SyncReq{}, client.SyncTimelineHas(roomID, func(result gjson.Result) bool {
-		if result.Get("type").Str == "m.room.member" {
-			eventID = result.Get("event_id").Str
-			return true
-		}
-		return false
-	}))
-	if eventID == "" {
-		t.Fatal("did not find an event_id")
-	}
+	roomID, eventID := createRoomForReadReceipts(t, alice, deployment)
 
 	reqBody := client.WithJSONBody(t, map[string]interface{}{
 		"m.fully_read": eventID,
 		"m.read":       eventID,
 	})
 	alice.MustDoFunc(t, "POST", []string{"_matrix", "client", "v3", "rooms", roomID, "read_markers"}, reqBody)
+
+	// Make sure the read receipt shows up in sync.
+	alice.MustSyncUntil(t, client.SyncReq{}, syncHasReadReceipt(roomID, alice.UserID, eventID))
+
+	// Make sure that the fully_read receipt shows up in account data via sync.
+	// Use the same token as above to replay the syncs.
+	alice.MustSyncUntil(t, client.SyncReq{}, client.SyncRoomAccountDataHas(roomID, func(result gjson.Result) bool {
+		return result.Get("type").Str == "m.fully_read" &&
+			result.Get("content.event_id").Str == eventID
+	}))
 }

tests/csapi/room_typing_test.go

@@ -1,7 +1,6 @@
 package csapi_tests
 
 import (
-	"fmt"
 	"testing"
 
 	"github.com/tidwall/gjson"
@@ -29,26 +28,15 @@ func TestTyping(t *testing.T) {
 		"timeout": 10000,
 	}))
 
-	bob.MustSyncUntil(t, client.SyncReq{Since: token}, func(clientUserID string, topLevelSyncJSON gjson.Result) error {
-		key := "rooms.join." + client.GjsonEscape(roomID) + ".ephemeral.events"
-		array := topLevelSyncJSON.Get(key)
-		if !array.Exists() {
-			return fmt.Errorf("Key %s does not exist", key)
+	bob.MustSyncUntil(t, client.SyncReq{Since: token}, client.SyncEphemeralHas(roomID, func(result gjson.Result) bool {
+		if result.Get("type").Str != "m.typing" {
+			return false
 		}
-		if !array.IsArray() {
-			return fmt.Errorf("Key %s exists but it isn't an array", key)
-		}
-		goArray := array.Array()
-		for _, ev := range goArray {
-			if ev.Get("type").Str != "m.typing" {
-				continue
-			}
-			for _, item := range ev.Get("content").Get("user_ids").Array() {
-				if item.Str == alice.UserID {
-					return nil
-				}
+		for _, item := range result.Get("content").Get("user_ids").Array() {
+			if item.Str == alice.UserID {
+				return true
 			}
 		}
-		return fmt.Errorf("no typing events")
-	})
+		return false
+	}))
 }