Refactor to allow passing actual filter

syncapi/storage/interface.go

@@ -64,10 +64,10 @@ type Database interface {
 	// from when the device sent the event via an API that included a transaction
 	// ID. A response object must be provided for IncrementaSync to populate - it
 	// will not create one.
-	IncrementalSync(ctx context.Context, res *types.Response, device userapi.Device, fromPos, toPos types.StreamingToken, numRecentEventsPerRoom int, wantFullState bool) (*types.Response, error)
+	IncrementalSync(ctx context.Context, res *types.Response, device userapi.Device, fromPos, toPos types.StreamingToken, filter *gomatrixserverlib.Filter, wantFullState bool) (*types.Response, error)
 	// CompleteSync returns a complete /sync API response for the given user. A response object
 	// must be provided for CompleteSync to populate - it will not create one.
-	CompleteSync(ctx context.Context, res *types.Response, device userapi.Device, numRecentEventsPerRoom int) (*types.Response, error)
+	CompleteSync(ctx context.Context, res *types.Response, device userapi.Device, filter *gomatrixserverlib.Filter) (*types.Response, error)
 	// GetAccountDataInRange returns all account data for a given user inserted or
 	// updated between two given positions
 	// Returns a map following the format data[roomID] = []dataTypes

syncapi/storage/shared/syncserver.go

@@ -685,7 +685,7 @@ func (d *Database) IncrementalSync(
 	ctx context.Context, res *types.Response,
 	device userapi.Device,
 	fromPos, toPos types.StreamingToken,
-	numRecentEventsPerRoom int,
+	filter *gomatrixserverlib.Filter,
 	wantFullState bool,
 ) (*types.Response, error) {
 	res.NextBatch = fromPos.WithUpdates(toPos)
@@ -697,6 +697,9 @@ func (d *Database) IncrementalSync(
 			From: fromPos.PDUPosition,
 			To:   toPos.PDUPosition,
 		}
+
+		numRecentEventsPerRoom := filter.Room.Timeline.Limit
+
 		joinedRoomIDs, err = d.addPDUDeltaToResponse(
 			ctx, device, r, numRecentEventsPerRoom, wantFullState, res,
 		)
@@ -761,8 +764,7 @@ func (d *Database) RedactEvent(ctx context.Context, redactedEventID string, reda
 func (d *Database) getResponseWithPDUsForCompleteSync(
 	ctx context.Context, res *types.Response,
 	userID string, device userapi.Device,
-	numRecentEventsPerRoom int,
+	filter *gomatrixserverlib.Filter,
-	includeLeave bool,
 ) (
 	toPos types.StreamingToken,
 	joinedRoomIDs []string,
@@ -801,13 +803,11 @@ func (d *Database) getResponseWithPDUsForCompleteSync(
 		return
 	}
 
-	stateFilter := gomatrixserverlib.DefaultStateFilter() // TODO: use filter provided in request
-
 	// Build up a /sync response. Add joined rooms.
 	for _, roomID := range joinedRoomIDs {
 		var jr *types.JoinResponse
 		jr, err = d.getJoinResponseForCompleteSync(
-			ctx, txn, roomID, r, &stateFilter, numRecentEventsPerRoom, device,
+			ctx, txn, roomID, r, filter, device,
 		)
 		if err != nil {
 			return
@@ -827,7 +827,7 @@ func (d *Database) getResponseWithPDUsForCompleteSync(
 		if !peek.Deleted {
 			var jr *types.JoinResponse
 			jr, err = d.getJoinResponseForCompleteSync(
-				ctx, txn, peek.RoomID, r, &stateFilter, numRecentEventsPerRoom, device,
+				ctx, txn, peek.RoomID, r, filter, device,
 			)
 			if err != nil {
 				return
@@ -848,14 +848,17 @@ func (d *Database) getJoinResponseForCompleteSync(
 	ctx context.Context, txn *sql.Tx,
 	roomID string,
 	r types.Range,
-	stateFilter *gomatrixserverlib.StateFilter,
+	filter *gomatrixserverlib.Filter,
-	numRecentEventsPerRoom int, device userapi.Device,
+	device userapi.Device,
 ) (jr *types.JoinResponse, err error) {
 	var stateEvents []*gomatrixserverlib.HeaderedEvent
-	stateEvents, err = d.CurrentRoomState.SelectCurrentState(ctx, txn, roomID, stateFilter)
+	stateEvents, err = d.CurrentRoomState.SelectCurrentState(ctx, txn, roomID, &filter.Room.State)
 	if err != nil {
 		return
 	}
+
+	numRecentEventsPerRoom := filter.Room.Timeline.Limit
+
 	// TODO: When filters are added, we may need to call this multiple times to get enough events.
 	//       See: https://github.com/matrix-org/synapse/blob/v0.19.3/synapse/handlers/sync.py#L316
 	var recentStreamEvents []types.StreamEvent
@@ -924,11 +927,13 @@ func (d *Database) getJoinResponseForCompleteSync(
 }
 
 func (d *Database) CompleteSync(
-	ctx context.Context, res *types.Response,
+	ctx context.Context,
-	device userapi.Device, numRecentEventsPerRoom int,
+	res *types.Response,
+	device userapi.Device,
+	filter *gomatrixserverlib.Filter,
 ) (*types.Response, error) {
 	toPos, joinedRoomIDs, err := d.getResponseWithPDUsForCompleteSync(
-		ctx, res, device.UserID, device, numRecentEventsPerRoom,
+		ctx, res, device.UserID, device, filter,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("d.getResponseWithPDUsForCompleteSync: %w", err)

syncapi/sync/request.go

@@ -33,21 +33,11 @@ const defaultSyncTimeout = time.Duration(0)
 const defaultIncludeLeave = false
 const DefaultTimelineLimit = 20
 
-type filter struct {
-	Room struct {
-		IncludeLeave *bool `json:"include_leave"`
-		Timeline     struct {
-			Limit *int `json:"limit"`
-		} `json:"timeline"`
-	} `json:"room"`
-}
-
 // syncRequest represents a /sync request, with sensible defaults/sanity checks applied.
 type syncRequest struct {
 	ctx           context.Context
 	device        userapi.Device
-	limit         int
+	filter        *gomatrixserverlib.Filter
-	includeLeave  bool
 	timeout       time.Duration
 	since         types.StreamingToken // nil means that no since token was supplied
 	wantFullState bool
@@ -66,21 +56,14 @@ func newSyncRequest(req *http.Request, device userapi.Device, syncDB storage.Dat
 			return nil, err
 		}
 	}
-	timelineLimit := DefaultTimelineLimit
+
-	includeLeave := defaultIncludeLeave
+	var f *gomatrixserverlib.Filter
 	// TODO: read from stored filters too
 	filterQuery := req.URL.Query().Get("filter")
 	if filterQuery != "" {
 		if filterQuery[0] == '{' {
 			// attempt to parse the timeline limit at least
-			var f filter
+			json.Unmarshal([]byte(filterQuery), &f)
-			err := json.Unmarshal([]byte(filterQuery), &f)
-			if err == nil && f.Room.Timeline.Limit != nil {
-				timelineLimit = *f.Room.Timeline.Limit
-			}
-			if err == nil && f.Room.IncludeLeave != nil {
-				includeLeave = *f.Room.IncludeLeave
-			}
 		} else {
 			// attempt to load the filter ID
 			localpart, _, err := gomatrixserverlib.SplitID('@', device.UserID)
@@ -88,10 +71,10 @@ func newSyncRequest(req *http.Request, device userapi.Device, syncDB storage.Dat
 				util.GetLogger(req.Context()).WithError(err).Error("gomatrixserverlib.SplitID failed")
 				return nil, err
 			}
-			f, err := syncDB.GetFilter(req.Context(), localpart, filterQuery)
+			f, err = syncDB.GetFilter(req.Context(), localpart, filterQuery)
-			if err == nil {
+			if err != nil {
-				timelineLimit = f.Room.Timeline.Limit
+				util.GetLogger(req.Context()).WithError(err).Error("syncDB.GetFilter failed")
-				includeLeave = f.Room.IncludeLeave
+				return nil, err
 			}
 		}
 	}
@@ -102,8 +85,7 @@ func newSyncRequest(req *http.Request, device userapi.Device, syncDB storage.Dat
 		timeout:       timeout,
 		since:         since,
 		wantFullState: wantFullState,
-		limit:         timelineLimit,
+		filter:        f,
-		includeLeave:  includeLeave,
 		log:           util.GetLogger(req.Context()),
 	}, nil
 }

syncapi/sync/requestpool.go

@@ -140,12 +140,11 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 	}
 
 	logger := util.GetLogger(req.Context()).WithFields(log.Fields{
-		"user_id":       device.UserID,
+		"user_id":   device.UserID,
-		"device_id":     device.ID,
+		"device_id": device.ID,
-		"since":         syncReq.since,
+		"since":     syncReq.since,
-		"timeout":       syncReq.timeout,
+		"timeout":   syncReq.timeout,
-		"limit":         syncReq.limit,
+		"filter":    syncReq.filter,
-		"include_leave": syncReq.includeLeave,
 	})
 
 	activeSyncRequests.Inc()
@@ -248,9 +247,12 @@ func (rp *RequestPool) OnIncomingKeyChangeRequest(req *http.Request, device *use
 			JSON: jsonerror.InvalidArgumentValue("bad 'to' value"),
 		}
 	}
+
 	// work out room joins/leaves
+	var f gomatrixserverlib.Filter
+	f.Room.Timeline.Limit = 10
 	res, err := rp.db.IncrementalSync(
-		req.Context(), types.NewResponse(), *device, fromToken, toToken, 10, false,
+		req.Context(), types.NewResponse(), *device, fromToken, toToken, &f, false,
 	)
 	if err != nil {
 		util.GetLogger(req.Context()).WithError(err).Error("Failed to IncrementalSync")
@@ -286,12 +288,12 @@ func (rp *RequestPool) currentSyncForUser(req syncRequest, latestPos types.Strea
 
 	// TODO: handle ignored users
 	if req.since.IsEmpty() {
-		res, err = rp.db.CompleteSync(req.ctx, res, req.device, req.limit, req.includeLeave)
+		res, err = rp.db.CompleteSync(req.ctx, res, req.device, req.filter)
 		if err != nil {
 			return res, fmt.Errorf("rp.db.CompleteSync: %w", err)
 		}
 	} else {
-		res, err = rp.db.IncrementalSync(req.ctx, res, req.device, req.since, latestPos, req.limit, req.wantFullState)
+		res, err = rp.db.IncrementalSync(req.ctx, res, req.device, req.since, latestPos, req.filter, req.wantFullState)
 		if err != nil {
 			return res, fmt.Errorf("rp.db.IncrementalSync: %w", err)
 		}