Use provided filter for account_data (#2372)

* Reuse IncrementalSync, use provided filter

* Inform SyncAPI about newly created push_rules

4 files changed, 10 insertions, 38 deletions

diff --git a/syncapi/streams/stream_accountdata.go b/syncapi/streams/stream_accountdata.go
index 105d8526..094c5148 100644
--- a/syncapi/streams/stream_accountdata.go
+++ b/syncapi/streams/stream_accountdata.go
@@ -30,37 +30,7 @@ func (p *AccountDataStreamProvider) CompleteSync(
 	ctx context.Context,
 	req *types.SyncRequest,
 ) types.StreamPosition {
-	dataReq := &userapi.QueryAccountDataRequest{
-		UserID: req.Device.UserID,
-	}
-	dataRes := &userapi.QueryAccountDataResponse{}
-	if err := p.userAPI.QueryAccountData(ctx, dataReq, dataRes); err != nil {
-		req.Log.WithError(err).Error("p.userAPI.QueryAccountData failed")
-		return p.LatestPosition(ctx)
-	}
-	for datatype, databody := range dataRes.GlobalAccountData {
-		req.Response.AccountData.Events = append(
-			req.Response.AccountData.Events,
-			gomatrixserverlib.ClientEvent{
-				Type:    datatype,
-				Content: gomatrixserverlib.RawJSON(databody),
-			},
-		)
-	}
-	for r, j := range req.Response.Rooms.Join {
-		for datatype, databody := range dataRes.RoomAccountData[r] {
-			j.AccountData.Events = append(
-				j.AccountData.Events,
-				gomatrixserverlib.ClientEvent{
-					Type:    datatype,
-					Content: gomatrixserverlib.RawJSON(databody),
-				},
-			)
-			req.Response.Rooms.Join[r] = j
-		}
-	}
-
-	return p.LatestPosition(ctx)
+	return p.IncrementalSync(ctx, req, 0, p.LatestPosition(ctx))
 }
 
 func (p *AccountDataStreamProvider) IncrementalSync(
@@ -72,10 +42,9 @@ func (p *AccountDataStreamProvider) IncrementalSync(
 		From: from,
 		To:   to,
 	}
-	accountDataFilter := gomatrixserverlib.DefaultEventFilter() // TODO: use filter provided in req instead
 
 	dataTypes, err := p.DB.GetAccountDataInRange(
-		ctx, req.Device.UserID, r, &accountDataFilter,
+		ctx, req.Device.UserID, r, &req.Filter.AccountData,
 	)
 	if err != nil {
 		req.Log.WithError(err).Error("p.DB.GetAccountDataInRange failed")
diff --git a/sytest-blacklist b/sytest-blacklist
index f1bd60db..713a5b63 100644
--- a/sytest-blacklist
+++ b/sytest-blacklist
@@ -1,7 +1,3 @@
-# Blacklisted until matrix-org/dendrite#862 is reverted due to Riot bug
-
-Latest account data appears in v2 /sync
-
 # Relies on a rejected PL event which will never be accepted into the DAG
 
 # Caused by <https://github.com/matrix-org/sytest/pull/911>
diff --git a/sytest-whitelist b/sytest-whitelist
index 979f12bf..2052185f 100644
--- a/sytest-whitelist
+++ b/sytest-whitelist
@@ -154,7 +154,7 @@ Can add account data
 Can add account data to room
 Can get account data without syncing
 Can get room account data without syncing
-#Latest account data appears in v2 /sync
+Latest account data appears in v2 /sync
 New account data appears in incremental v2 /sync
 Checking local federation server
 Inbound federation can query profile data
diff --git a/userapi/internal/api.go b/userapi/internal/api.go
index d1c12f05..be58e2d8 100644
--- a/userapi/internal/api.go
+++ b/userapi/internal/api.go
@@ -90,6 +90,13 @@ func (a *UserInternalAPI) PerformAccountCreation(ctx context.Context, req *api.P
 		return nil
 	}
 
+	// Inform the SyncAPI about the newly created push_rules
+	if err = a.SyncProducer.SendAccountData(acc.UserID, "", "m.push_rules"); err != nil {
+		util.GetLogger(ctx).WithFields(logrus.Fields{
+			"user_id": acc.UserID,
+		}).WithError(err).Warn("failed to send account data to the SyncAPI")
+	}
+
 	if req.AccountType == api.AccountTypeGuest {
 		res.AccountCreated = true
 		res.Account = acc