mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/element-hq/dendrite.git synced 2025-09-14 13:22:26 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

State events filtering database api (#438)

Browse source 
This PR adds a gomatrixserverlib.Filter parameter to functions handling the syncapi_current_room_state table. It does not implement any filtering logic inside the syncapi IncrementalSync/CompleteSync functions, just the APIs for future use.

Default filters are provided as placeholders in IncrementalSync/CompleteSync, so behaviour should be unchanged (except the default 20 event limit)

SQL table will be changed. You can upgrade an existing database using:

```
ALTER TABLE syncapi_current_room_state  ADD COLUMN IF NOT EXISTS sender text;
UPDATE syncapi_current_room_state SET sender=(event_json::json->>'sender');
ALTER TABLE syncapi_current_room_state ALTER COLUMN sender SET NOT NULL;
ALTER TABLE syncapi_current_room_state  ADD COLUMN IF NOT EXISTS contains_url bool;
UPDATE syncapi_current_room_state SET contains_url=(event_json::json->>'content')::json->>'url' IS NOT NULL;
ALTER TABLE syncapi_current_room_state ALTER COLUMN contains_url SET NOT NULL;
```

Note: This depends on #436 (and includes all its commits). I'm not sure if Github will remove the duplicated commits once #436 is merged.
This commit is contained in:
Thibaut CHARLES 2019-08-07 12:12:09 +02:00 • committed by Andrew Morgan
parent 94ea325c93
commit 76e4ebaf78
5 changed files with 134 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

31
syncapi/storage/syncserver.go
View file

@ -185,10 +185,10 @@ func (d *SyncServerDatasource) 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 *SyncServerDatasource) GetStateEventsForRoom(
ctx context.Context, roomID string,
ctx context.Context, roomID string, stateFilterPart *gomatrixserverlib.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
@ -245,6 +245,8 @@ func (d *SyncServerDatasource) addPDUDeltaToResponse(
var succeeded bool
defer common.EndTransaction(txn, &succeeded)
stateFilterPart := gomatrixserverlib.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 PDU stream positions.
// This works out what the 'state' key should be for each room as well as which membership block
@ -252,9 +254,13 @@ func (d *SyncServerDatasource) addPDUDeltaToResponse(
var deltas []stateDelta
var joinedRoomIDs []string
if !wantFullState {
deltas, joinedRoomIDs, err = d.getStateDeltas(ctx, &device, txn, fromPos, toPos, device.UserID)
deltas, joinedRoomIDs, err = d.getStateDeltas(
ctx, &device, txn, fromPos, toPos, device.UserID, &stateFilterPart,
)
} else {
deltas, joinedRoomIDs, err = d.getStateDeltasForFullStateSync(ctx, &device, txn, fromPos, toPos, device.UserID)
deltas, joinedRoomIDs, err = d.getStateDeltasForFullStateSync(
ctx, &device, txn, fromPos, toPos, device.UserID, &stateFilterPart,
)
}
if err != nil {
return nil, err
@ -404,10 +410,12 @@ func (d *SyncServerDatasource) getResponseWithPDUsForCompleteSync(
return
}
stateFilterPart := gomatrixserverlib.DefaultFilterPart() // TODO: use filter provided in request
// Build up a /sync response. Add joined rooms.
for _, roomID := range joinedRoomIDs {
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
}
@ -733,6 +741,7 @@ func (d *SyncServerDatasource) fetchMissingStateEvents(
func (d *SyncServerDatasource) getStateDeltas(
ctx context.Context, device *authtypes.Device, txn *sql.Tx,
fromPos, toPos int64, userID string,
stateFilterPart *gomatrixserverlib.FilterPart,
) ([]stateDelta, []string, 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
@ -745,7 +754,7 @@ func (d *SyncServerDatasource) getStateDeltas(
var deltas []stateDelta
// get all the state events ever between these two positions
stateNeeded, eventMap, err := d.events.selectStateInRange(ctx, txn, fromPos, toPos)
stateNeeded, eventMap, err := d.events.selectStateInRange(ctx, txn, fromPos, toPos, stateFilterPart)
if err != nil {
return nil, nil, err
}
@ -765,7 +774,7 @@ func (d *SyncServerDatasource) getStateDeltas(
if membership == gomatrixserverlib.Join {
// send full room state down instead of a delta
var s []streamEvent
s, err = d.currentStateStreamEventsForRoom(ctx, txn, roomID)
s, err = d.currentStateStreamEventsForRoom(ctx, txn, roomID, stateFilterPart)
if err != nil {
return nil, nil, err
}
@ -807,6 +816,7 @@ func (d *SyncServerDatasource) getStateDeltas(
func (d *SyncServerDatasource) getStateDeltasForFullStateSync(
ctx context.Context, device *authtypes.Device, txn *sql.Tx,
fromPos, toPos int64, userID string,
stateFilterPart *gomatrixserverlib.FilterPart,
) ([]stateDelta, []string, error) {
joinedRoomIDs, err := d.roomstate.selectRoomIDsWithMembership(ctx, txn, userID, gomatrixserverlib.Join)
if err != nil {
@ -818,7 +828,7 @@ func (d *SyncServerDatasource) getStateDeltasForFullStateSync(
// Add full states for all joined rooms
for _, joinedRoomID := range joinedRoomIDs {
s, stateErr := d.currentStateStreamEventsForRoom(ctx, txn, joinedRoomID)
s, stateErr := d.currentStateStreamEventsForRoom(ctx, txn, joinedRoomID, stateFilterPart)
if stateErr != nil {
return nil, nil, stateErr
}
@ -830,7 +840,7 @@ func (d *SyncServerDatasource) getStateDeltasForFullStateSync(
}
// Get all the state events ever between these two positions
stateNeeded, eventMap, err := d.events.selectStateInRange(ctx, txn, fromPos, toPos)
stateNeeded, eventMap, err := d.events.selectStateInRange(ctx, txn, fromPos, toPos, stateFilterPart)
if err != nil {
return nil, nil, err
}
@ -861,8 +871,9 @@ func (d *SyncServerDatasource) getStateDeltasForFullStateSync(
func (d *SyncServerDatasource) currentStateStreamEventsForRoom(
ctx context.Context, txn *sql.Tx, roomID string,
stateFilterPart *gomatrixserverlib.FilterPart,
) ([]streamEvent, error) {
allState, err := d.roomstate.selectCurrentState(ctx, txn, roomID)
allState, err := d.roomstate.selectCurrentState(ctx, txn, roomID, stateFilterPart)
if err != nil {
return nil, err
}