From c922cdce324283641120844f3800d938366588cc Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Tue, 19 Jan 2021 09:37:35 +0000 Subject: [PATCH] Add some filtering (postgres only for now) --- syncapi/storage/interface.go | 4 +-- .../postgres/output_room_events_table.go | 25 +++++++++++++++---- syncapi/storage/shared/syncserver.go | 10 ++++---- .../sqlite3/output_room_events_table.go | 9 ++++--- syncapi/storage/tables/interface.go | 2 +- syncapi/streams/stream_pdu.go | 19 ++++++++------ 6 files changed, 46 insertions(+), 23 deletions(-) diff --git a/syncapi/storage/interface.go b/syncapi/storage/interface.go index a51ab4e0d..9d9c9fad7 100644 --- a/syncapi/storage/interface.go +++ b/syncapi/storage/interface.go @@ -40,7 +40,7 @@ type Database interface { GetStateDeltas(ctx context.Context, device *userapi.Device, r types.Range, userID string, stateFilter *gomatrixserverlib.StateFilter) ([]types.StateDelta, []string, error) RoomIDsWithMembership(ctx context.Context, userID string, membership string) ([]string, error) - RecentEvents(ctx context.Context, roomID string, r types.Range, limit int, chronologicalOrder bool, onlySyncEvents bool) ([]types.StreamEvent, bool, error) + RecentEvents(ctx context.Context, roomID string, r types.Range, eventFilter *gomatrixserverlib.EventFilter, chronologicalOrder bool, onlySyncEvents bool) ([]types.StreamEvent, bool, error) GetBackwardTopologyPos(ctx context.Context, events []types.StreamEvent) (types.TopologyToken, error) PositionInTopology(ctx context.Context, eventID string) (pos types.StreamPosition, spos types.StreamPosition, err error) @@ -105,7 +105,7 @@ type Database interface { // Returns an error if there was a problem communicating with the database. DeletePeeks(ctx context.Context, RoomID, UserID string) (types.StreamPosition, error) // GetEventsInStreamingRange retrieves all of the events on a given ordering using the given extremities and limit. - GetEventsInStreamingRange(ctx context.Context, from, to *types.StreamingToken, roomID string, limit int, backwardOrdering bool) (events []types.StreamEvent, err error) + GetEventsInStreamingRange(ctx context.Context, from, to *types.StreamingToken, roomID string, eventFilter *gomatrixserverlib.EventFilter, backwardOrdering bool) (events []types.StreamEvent, err error) // GetEventsInTopologicalRange retrieves all of the events on a given ordering using the given extremities and limit. GetEventsInTopologicalRange(ctx context.Context, from, to *types.TopologyToken, roomID string, limit int, backwardOrdering bool) (events []types.StreamEvent, err error) // EventPositionInTopology returns the depth and stream position of the given event. diff --git a/syncapi/storage/postgres/output_room_events_table.go b/syncapi/storage/postgres/output_room_events_table.go index f4bbebd26..c48cc6fbb 100644 --- a/syncapi/storage/postgres/output_room_events_table.go +++ b/syncapi/storage/postgres/output_room_events_table.go @@ -84,12 +84,20 @@ const selectEventsSQL = "" + const selectRecentEventsSQL = "" + "SELECT event_id, id, headered_event_json, session_id, exclude_from_sync, transaction_id FROM syncapi_output_room_events" + " WHERE room_id = $1 AND id > $2 AND id <= $3" + - " ORDER BY id DESC LIMIT $4" + " AND ( $4::text[] IS NULL OR sender = ANY($4) )" + + " AND ( $5::text[] IS NULL OR NOT(sender = ANY($5)) )" + + " AND ( $6::text[] IS NULL OR type LIKE ANY($6) )" + + " AND ( $7::text[] IS NULL OR NOT(type LIKE ANY($7)) )" + + " ORDER BY id DESC LIMIT $8" const selectRecentEventsForSyncSQL = "" + "SELECT event_id, id, headered_event_json, session_id, exclude_from_sync, transaction_id FROM syncapi_output_room_events" + " WHERE room_id = $1 AND id > $2 AND id <= $3 AND exclude_from_sync = FALSE" + - " ORDER BY id DESC LIMIT $4" + " AND ( $4::text[] IS NULL OR sender = ANY($4) )" + + " AND ( $5::text[] IS NULL OR NOT(sender = ANY($5)) )" + + " AND ( $6::text[] IS NULL OR type LIKE ANY($6) )" + + " AND ( $7::text[] IS NULL OR NOT(type LIKE ANY($7)) )" + + " ORDER BY id DESC LIMIT $8" const selectEarlyEventsSQL = "" + "SELECT event_id, id, headered_event_json, session_id, exclude_from_sync, transaction_id FROM syncapi_output_room_events" + @@ -322,7 +330,7 @@ func (s *outputRoomEventsStatements) InsertEvent( // from sync. func (s *outputRoomEventsStatements) SelectRecentEvents( ctx context.Context, txn *sql.Tx, - roomID string, r types.Range, limit int, + roomID string, r types.Range, eventFilter *gomatrixserverlib.EventFilter, chronologicalOrder bool, onlySyncEvents bool, ) ([]types.StreamEvent, bool, error) { var stmt *sql.Stmt @@ -331,7 +339,14 @@ func (s *outputRoomEventsStatements) SelectRecentEvents( } else { stmt = sqlutil.TxStmt(txn, s.selectRecentEventsStmt) } - rows, err := stmt.QueryContext(ctx, roomID, r.Low(), r.High(), limit+1) + rows, err := stmt.QueryContext( + ctx, roomID, r.Low(), r.High(), + pq.StringArray(eventFilter.Senders), + pq.StringArray(eventFilter.NotSenders), + pq.StringArray(filterConvertTypeWildcardToSQL(eventFilter.Types)), + pq.StringArray(filterConvertTypeWildcardToSQL(eventFilter.NotTypes)), + eventFilter.Limit+1, + ) if err != nil { return nil, false, err } @@ -350,7 +365,7 @@ func (s *outputRoomEventsStatements) SelectRecentEvents( } // we queried for 1 more than the limit, so if we returned one more mark limited=true limited := false - if len(events) > limit { + if len(events) > eventFilter.Limit { limited = true // re-slice the extra (oldest) event out: in chronological order this is the first entry, else the last. if chronologicalOrder { diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go index 5b06aabcd..555b17043 100644 --- a/syncapi/storage/shared/syncserver.go +++ b/syncapi/storage/shared/syncserver.go @@ -110,8 +110,8 @@ func (d *Database) RoomIDsWithMembership(ctx context.Context, userID string, mem return d.CurrentRoomState.SelectRoomIDsWithMembership(ctx, nil, userID, membership) } -func (d *Database) RecentEvents(ctx context.Context, roomID string, r types.Range, limit int, chronologicalOrder bool, onlySyncEvents bool) ([]types.StreamEvent, bool, error) { - return d.OutputEvents.SelectRecentEvents(ctx, nil, roomID, r, limit, chronologicalOrder, onlySyncEvents) +func (d *Database) RecentEvents(ctx context.Context, roomID string, r types.Range, eventFilter *gomatrixserverlib.EventFilter, chronologicalOrder bool, onlySyncEvents bool) ([]types.StreamEvent, bool, error) { + return d.OutputEvents.SelectRecentEvents(ctx, nil, roomID, r, eventFilter, chronologicalOrder, onlySyncEvents) } func (d *Database) PositionInTopology(ctx context.Context, eventID string) (pos types.StreamPosition, spos types.StreamPosition, err error) { @@ -151,7 +151,7 @@ func (d *Database) Events(ctx context.Context, eventIDs []string) ([]*gomatrixse func (d *Database) GetEventsInStreamingRange( ctx context.Context, from, to *types.StreamingToken, - roomID string, limit int, + roomID string, eventFilter *gomatrixserverlib.EventFilter, backwardOrdering bool, ) (events []types.StreamEvent, err error) { r := types.Range{ @@ -162,14 +162,14 @@ func (d *Database) GetEventsInStreamingRange( if backwardOrdering { // When using backward ordering, we want the most recent events first. if events, _, err = d.OutputEvents.SelectRecentEvents( - ctx, nil, roomID, r, limit, false, false, + ctx, nil, roomID, r, eventFilter, false, false, ); err != nil { return } } else { // When using forward ordering, we want the least recent events first. if events, err = d.OutputEvents.SelectEarlyEvents( - ctx, nil, roomID, r, limit, + ctx, nil, roomID, r, eventFilter.Limit, // TODO: filter here too ); err != nil { return } diff --git a/syncapi/storage/sqlite3/output_room_events_table.go b/syncapi/storage/sqlite3/output_room_events_table.go index edbd36fb1..7218cf420 100644 --- a/syncapi/storage/sqlite3/output_room_events_table.go +++ b/syncapi/storage/sqlite3/output_room_events_table.go @@ -175,6 +175,9 @@ func (s *outputRoomEventsStatements) SelectStateInRange( ctx context.Context, txn *sql.Tx, r types.Range, stateFilterPart *gomatrixserverlib.StateFilter, ) (map[string]map[string]bool, map[string]types.StreamEvent, error) { + params := []interface{}{} + _ = params + stmt := sqlutil.TxStmt(txn, s.selectStateInRangeStmt) rows, err := stmt.QueryContext( @@ -333,7 +336,7 @@ func (s *outputRoomEventsStatements) InsertEvent( func (s *outputRoomEventsStatements) SelectRecentEvents( ctx context.Context, txn *sql.Tx, - roomID string, r types.Range, limit int, + roomID string, r types.Range, eventFilter *gomatrixserverlib.EventFilter, chronologicalOrder bool, onlySyncEvents bool, ) ([]types.StreamEvent, bool, error) { var stmt *sql.Stmt @@ -343,7 +346,7 @@ func (s *outputRoomEventsStatements) SelectRecentEvents( stmt = sqlutil.TxStmt(txn, s.selectRecentEventsStmt) } - rows, err := stmt.QueryContext(ctx, roomID, r.Low(), r.High(), limit+1) + rows, err := stmt.QueryContext(ctx, roomID, r.Low(), r.High(), eventFilter.Limit+1) if err != nil { return nil, false, err } @@ -362,7 +365,7 @@ func (s *outputRoomEventsStatements) SelectRecentEvents( } // we queried for 1 more than the limit, so if we returned one more mark limited=true limited := false - if len(events) > limit { + if len(events) > eventFilter.Limit { limited = true // re-slice the extra (oldest) event out: in chronological order this is the first entry, else the last. if chronologicalOrder { diff --git a/syncapi/storage/tables/interface.go b/syncapi/storage/tables/interface.go index fca888249..2ee4fb469 100644 --- a/syncapi/storage/tables/interface.go +++ b/syncapi/storage/tables/interface.go @@ -56,7 +56,7 @@ type Events interface { // SelectRecentEvents returns events between the two stream positions: exclusive of low and inclusive of high. // If onlySyncEvents has a value of true, only returns the events that aren't marked as to exclude from sync. // Returns up to `limit` events. Returns `limited=true` if there are more events in this range but we hit the `limit`. - SelectRecentEvents(ctx context.Context, txn *sql.Tx, roomID string, r types.Range, limit int, chronologicalOrder bool, onlySyncEvents bool) ([]types.StreamEvent, bool, error) + SelectRecentEvents(ctx context.Context, txn *sql.Tx, roomID string, r types.Range, eventFilter *gomatrixserverlib.EventFilter, chronologicalOrder bool, onlySyncEvents bool) ([]types.StreamEvent, bool, error) // SelectEarlyEvents returns the earliest events in the given room. SelectEarlyEvents(ctx context.Context, txn *sql.Tx, roomID string, r types.Range, limit int) ([]types.StreamEvent, error) SelectEvents(ctx context.Context, txn *sql.Tx, eventIDs []string) ([]types.StreamEvent, error) diff --git a/syncapi/streams/stream_pdu.go b/syncapi/streams/stream_pdu.go index 483be575e..c81937f35 100644 --- a/syncapi/streams/stream_pdu.go +++ b/syncapi/streams/stream_pdu.go @@ -49,12 +49,14 @@ func (p *PDUStreamProvider) CompleteSync( } stateFilter := gomatrixserverlib.DefaultStateFilter() // TODO: use filter provided in request + eventFilter := gomatrixserverlib.DefaultEventFilter() + eventFilter.Limit = req.Limit // Build up a /sync response. Add joined rooms. for _, roomID := range joinedRoomIDs { var jr *types.JoinResponse jr, err = p.getJoinResponseForCompleteSync( - ctx, roomID, r, &stateFilter, req.Limit, req.Device, + ctx, roomID, r, &stateFilter, &eventFilter, req.Device, ) if err != nil { req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed") @@ -74,7 +76,7 @@ func (p *PDUStreamProvider) CompleteSync( if !peek.Deleted { var jr *types.JoinResponse jr, err = p.getJoinResponseForCompleteSync( - ctx, peek.RoomID, r, &stateFilter, req.Limit, req.Device, + ctx, peek.RoomID, r, &stateFilter, &eventFilter, req.Device, ) if err != nil { req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed") @@ -106,6 +108,8 @@ func (p *PDUStreamProvider) IncrementalSync( // TODO: use filter provided in request stateFilter := gomatrixserverlib.DefaultStateFilter() + eventFilter := gomatrixserverlib.DefaultEventFilter() + eventFilter.Limit = req.Limit if req.WantFullState { if stateDeltas, joinedRooms, err = p.DB.GetStateDeltasForFullStateSync(ctx, req.Device, r, req.Device.UserID, &stateFilter); err != nil { @@ -124,7 +128,7 @@ func (p *PDUStreamProvider) IncrementalSync( } for _, delta := range stateDeltas { - if err = p.addRoomDeltaToResponse(ctx, req.Device, r, delta, req.Limit, req.Response); err != nil { + if err = p.addRoomDeltaToResponse(ctx, req.Device, r, delta, &eventFilter, req.Response); err != nil { req.Log.WithError(err).Error("d.addRoomDeltaToResponse failed") return newPos } @@ -138,7 +142,7 @@ func (p *PDUStreamProvider) addRoomDeltaToResponse( device *userapi.Device, r types.Range, delta types.StateDelta, - numRecentEventsPerRoom int, + eventFilter *gomatrixserverlib.EventFilter, res *types.Response, ) error { if delta.MembershipPos > 0 && delta.Membership == gomatrixserverlib.Leave { @@ -152,7 +156,7 @@ func (p *PDUStreamProvider) addRoomDeltaToResponse( } recentStreamEvents, limited, err := p.DB.RecentEvents( ctx, delta.RoomID, r, - numRecentEventsPerRoom, true, true, + eventFilter, true, true, ) if err != nil { return err @@ -209,7 +213,8 @@ func (p *PDUStreamProvider) getJoinResponseForCompleteSync( roomID string, r types.Range, stateFilter *gomatrixserverlib.StateFilter, - numRecentEventsPerRoom int, device *userapi.Device, + eventFilter *gomatrixserverlib.EventFilter, + device *userapi.Device, ) (jr *types.JoinResponse, err error) { var stateEvents []*gomatrixserverlib.HeaderedEvent stateEvents, err = p.DB.CurrentState(ctx, roomID, stateFilter) @@ -221,7 +226,7 @@ func (p *PDUStreamProvider) getJoinResponseForCompleteSync( var recentStreamEvents []types.StreamEvent var limited bool recentStreamEvents, limited, err = p.DB.RecentEvents( - ctx, roomID, r, numRecentEventsPerRoom, true, true, + ctx, roomID, r, eventFilter, true, true, ) if err != nil { return