Impl filtering ability in syncapi_current_room_state SQL queries

Signed-off-by: Thibaut CHARLES cromfr@gmail.com

src/github.com/matrix-org/dendrite/syncapi/routing/state.go

@@ -22,6 +22,7 @@ import (
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
 	"github.com/matrix-org/dendrite/syncapi/storage"
 	"github.com/matrix-org/dendrite/syncapi/types"
+	"github.com/matrix-org/gomatrix"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/matrix-org/util"
 	log "github.com/sirupsen/logrus"
@@ -44,7 +45,10 @@ func OnIncomingStateRequest(req *http.Request, db *storage.SyncServerDatabase, r
 	// TODO(#287): Auth request and handle the case where the user has left (where
 	// we should return the state at the poin they left)
 
-	stateEvents, err := db.GetStateEventsForRoom(req.Context(), roomID)
+	stateFilterPart := gomatrix.DefaultFilterPart()
+	// TODO: stateFilterPart should not limit the number of state events (or only limits abusive number of events)
+
+	stateEvents, err := db.GetStateEventsForRoom(req.Context(), roomID, &stateFilterPart)
 	if err != nil {
 		return httputil.LogThenError(req, err)
 	}

src/github.com/matrix-org/dendrite/syncapi/storage/current_room_state_table.go

@@ -17,9 +17,11 @@ package storage
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 
 	"github.com/lib/pq"
 	"github.com/matrix-org/dendrite/common"
+	"github.com/matrix-org/gomatrix"
 	"github.com/matrix-org/gomatrixserverlib"
 )
 
@@ -32,6 +34,10 @@ CREATE TABLE IF NOT EXISTS syncapi_current_room_state (
     event_id TEXT NOT NULL,
     -- The state event type e.g 'm.room.member'
     type TEXT NOT NULL,
+    -- The 'sender' property of the event.
+    sender TEXT NOT NULL,
+	-- true if the event content contains a url key
+    contains_url BOOL NOT NULL,
     -- The state_key value for this state event e.g ''
     state_key TEXT NOT NULL,
     -- The JSON for the event. Stored as TEXT because this should be valid UTF-8.
@@ -46,16 +52,16 @@ CREATE TABLE IF NOT EXISTS syncapi_current_room_state (
     CONSTRAINT syncapi_room_state_unique UNIQUE (room_id, type, state_key)
 );
 -- for event deletion
-CREATE UNIQUE INDEX IF NOT EXISTS syncapi_event_id_idx ON syncapi_current_room_state(event_id);
+CREATE UNIQUE INDEX IF NOT EXISTS syncapi_event_id_idx ON syncapi_current_room_state(event_id, room_id, type, sender, contains_url);
 -- for querying membership states of users
 CREATE INDEX IF NOT EXISTS syncapi_membership_idx ON syncapi_current_room_state(type, state_key, membership) WHERE membership IS NOT NULL AND membership != 'leave';
 `
 
 const upsertRoomStateSQL = "" +
-	"INSERT INTO syncapi_current_room_state (room_id, event_id, type, state_key, event_json, membership, added_at)" +
-	" VALUES ($1, $2, $3, $4, $5, $6, $7)" +
+	"INSERT INTO syncapi_current_room_state (room_id, event_id, type, sender, contains_url, state_key, event_json, membership, added_at)" +
+	" VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)" +
 	" ON CONFLICT ON CONSTRAINT syncapi_room_state_unique" +
-	" DO UPDATE SET event_id = $2, event_json = $5, membership = $6, added_at = $7"
+	" DO UPDATE SET event_id = $2, sender=$4, contains_url=$5, event_json = $7, membership = $8, added_at = $9"
 
 const deleteRoomStateByEventIDSQL = "" +
 	"DELETE FROM syncapi_current_room_state WHERE event_id = $1"
@@ -64,7 +70,13 @@ const selectRoomIDsWithMembershipSQL = "" +
 	"SELECT room_id FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = $1 AND membership = $2"
 
 const selectCurrentStateSQL = "" +
-	"SELECT event_json FROM syncapi_current_room_state WHERE room_id = $1"
+	"SELECT event_json FROM syncapi_current_room_state WHERE room_id = $1" +
+	" AND ( $2::text[] IS NULL OR     sender  = ANY($2)  )" +
+	" AND ( $3::text[] IS NULL OR NOT(sender  = ANY($3)) )" +
+	" AND ( $4::text[] IS NULL OR     type LIKE ANY($4)  )" +
+	" AND ( $5::text[] IS NULL OR NOT(type LIKE ANY($5)) )" +
+	" AND ( $6::bool IS NULL   OR     contains_url = $6  )" +
+	" LIMIT $7"
 
 const selectJoinedUsersSQL = "" +
 	"SELECT room_id, state_key FROM syncapi_current_room_state WHERE type = 'm.room.member' AND membership = 'join'"
@@ -166,9 +178,17 @@ func (s *currentRoomStateStatements) selectRoomIDsWithMembership(
 // CurrentState returns all the current state events for the given room.
 func (s *currentRoomStateStatements) selectCurrentState(
 	ctx context.Context, txn *sql.Tx, roomID string,
+	stateFilterPart *gomatrix.FilterPart,
 ) ([]gomatrixserverlib.Event, error) {
 	stmt := common.TxStmt(txn, s.selectCurrentStateStmt)
-	rows, err := stmt.QueryContext(ctx, roomID)
+	rows, err := stmt.QueryContext(ctx, roomID,
+		pq.StringArray(stateFilterPart.Senders),
+		pq.StringArray(stateFilterPart.NotSenders),
+		pq.StringArray(filterConvertTypeWildcardToSQL(stateFilterPart.Types)),
+		pq.StringArray(filterConvertTypeWildcardToSQL(stateFilterPart.NotTypes)),
+		stateFilterPart.ContainsURL,
+		stateFilterPart.Limit,
+	)
 	if err != nil {
 		return nil, err
 	}
@@ -189,12 +209,23 @@ func (s *currentRoomStateStatements) upsertRoomState(
 	ctx context.Context, txn *sql.Tx,
 	event gomatrixserverlib.Event, membership *string, addedAt int64,
 ) error {
+	// Parse content as JSON and search for an "url" key
+	containsURL := false
+	var content map[string]interface{}
+	if json.Unmarshal(event.Content(), &content) != nil {
+		// Set containsURL = true if url is present
+		_, containsURL = content["url"]
+	}
+
+	// upsert state event
 	stmt := common.TxStmt(txn, s.upsertRoomStateStmt)
 	_, err := stmt.ExecContext(
 		ctx,
 		event.RoomID(),
 		event.EventID(),
 		event.Type(),
+		event.Sender(),
+		containsURL,
 		*event.StateKey(),
 		event.JSON(),
 		membership,

src/github.com/matrix-org/dendrite/syncapi/storage/filtering.go

@@ -0,0 +1,30 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package storage
+
+import (
+	"strings"
+)
+
+// filterConvertWildcardToSQL converts wildcards as defined in
+// https://matrix.org/docs/spec/client_server/r0.3.0.html#post-matrix-client-r0-user-userid-filter
+// to SQL wildcards that can be used with LIKE()
+func filterConvertTypeWildcardToSQL(values []string) []string {
+	ret := make([]string, len(values))
+	for i := range values {
+		ret[i] = strings.Replace(values[i], "*", "%", -1)
+	}
+	return ret
+}

src/github.com/matrix-org/dendrite/syncapi/storage/syncserver.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
 	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/gomatrix"
 	// Import the postgres database driver.
 	_ "github.com/lib/pq"
 	"github.com/matrix-org/dendrite/common"
@@ -177,10 +178,10 @@ func (d *SyncServerDatabase) GetStateEvent(
 // Returns an empty slice if no state events could be found for this room.
 // Returns an error if there was an issue with the retrieval.
 func (d *SyncServerDatabase) GetStateEventsForRoom(
-	ctx context.Context, roomID string,
+	ctx context.Context, roomID string, stateFilterPart *gomatrix.FilterPart,
 ) (stateEvents []gomatrixserverlib.Event, err error) {
 	err = common.WithTransaction(d.db, func(txn *sql.Tx) error {
-		stateEvents, err = d.roomstate.selectCurrentState(ctx, txn, roomID)
+		stateEvents, err = d.roomstate.selectCurrentState(ctx, txn, roomID, stateFilterPart)
 		return err
 	})
 	return
@@ -233,11 +234,13 @@ func (d *SyncServerDatabase) IncrementalSync(
 	var succeeded bool
 	defer common.EndTransaction(txn, &succeeded)
 
+	stateFilterPart := gomatrix.DefaultFilterPart() // TODO: use filter provided in request
+
 	// Work out which rooms to return in the response. This is done by getting not only the currently
 	// joined rooms, but also which rooms have membership transitions for this user between the 2 stream positions.
 	// This works out what the 'state' key should be for each room as well as which membership block
 	// to put the room into.
-	deltas, err := d.getStateDeltas(ctx, &device, txn, fromPos, toPos, device.UserID)
+	deltas, err := d.getStateDeltas(ctx, &device, txn, fromPos, toPos, device.UserID, &stateFilterPart)
 	if err != nil {
 		return nil, err
 	}
@@ -286,11 +289,13 @@ func (d *SyncServerDatabase) CompleteSync(
 		return nil, err
 	}
 
+	stateFilterPart := gomatrix.DefaultFilterPart() // TODO: use filter provided in request
+
 	// Build up a /sync response. Add joined rooms.
 	res := types.NewResponse(pos)
 	for _, roomID := range roomIDs {
 		var stateEvents []gomatrixserverlib.Event
-		stateEvents, err = d.roomstate.selectCurrentState(ctx, txn, roomID)
+		stateEvents, err = d.roomstate.selectCurrentState(ctx, txn, roomID, &stateFilterPart)
 		if err != nil {
 			return nil, err
 		}
@@ -547,6 +552,7 @@ func (d *SyncServerDatabase) fetchMissingStateEvents(
 func (d *SyncServerDatabase) getStateDeltas(
 	ctx context.Context, device *authtypes.Device, txn *sql.Tx,
 	fromPos, toPos types.StreamPosition, userID string,
+	stateFilterPart *gomatrix.FilterPart,
 ) ([]stateDelta, error) {
 	// Implement membership change algorithm: https://github.com/matrix-org/synapse/blob/v0.19.3/synapse/handlers/sync.py#L821
 	// - Get membership list changes for this user in this sync response
@@ -579,7 +585,7 @@ func (d *SyncServerDatabase) getStateDeltas(
 				if membership == "join" {
 					// send full room state down instead of a delta
 					var allState []gomatrixserverlib.Event
-					allState, err = d.roomstate.selectCurrentState(ctx, txn, roomID)
+					allState, err = d.roomstate.selectCurrentState(ctx, txn, roomID, stateFilterPart)
 					if err != nil {
 						return nil, err
 					}