mirror of
https://github.com/matrix-org/dendrite.git
synced 2025-12-12 17:33:09 -06:00
parent
2c3e55c7d3
commit
dd65a8fe68
|
|
@ -44,7 +44,7 @@ func OnIncomingStateRequest(req *http.Request, db *storage.SyncServerDatabase, r
|
||||||
// TODO(#287): Auth request and handle the case where the user has left (where
|
// TODO(#287): Auth request and handle the case where the user has left (where
|
||||||
// we should return the state at the poin they left)
|
// we should return the state at the poin they left)
|
||||||
|
|
||||||
stateEvents, err := db.GetStateEventsForRoom(req.Context(), roomID)
|
stateEvents, err := db.GetStateEventsForRoom(req.Context(), roomID, nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return httputil.LogThenError(req, err)
|
return httputil.LogThenError(req, err)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -20,6 +20,7 @@ import (
|
||||||
|
|
||||||
"github.com/lib/pq"
|
"github.com/lib/pq"
|
||||||
"github.com/matrix-org/dendrite/common"
|
"github.com/matrix-org/dendrite/common"
|
||||||
|
"github.com/matrix-org/gomatrix"
|
||||||
"github.com/matrix-org/gomatrixserverlib"
|
"github.com/matrix-org/gomatrixserverlib"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -32,6 +33,8 @@ CREATE TABLE IF NOT EXISTS syncapi_current_room_state (
|
||||||
event_id TEXT NOT NULL,
|
event_id TEXT NOT NULL,
|
||||||
-- The state event type e.g 'm.room.member'
|
-- The state event type e.g 'm.room.member'
|
||||||
type TEXT NOT NULL,
|
type TEXT NOT NULL,
|
||||||
|
-- The 'sender' property for the event.
|
||||||
|
sender TEXT NOT NULL,
|
||||||
-- The state_key value for this state event e.g ''
|
-- The state_key value for this state event e.g ''
|
||||||
state_key TEXT NOT NULL,
|
state_key TEXT NOT NULL,
|
||||||
-- The JSON for the event. Stored as TEXT because this should be valid UTF-8.
|
-- The JSON for the event. Stored as TEXT because this should be valid UTF-8.
|
||||||
|
|
@ -46,16 +49,17 @@ CREATE TABLE IF NOT EXISTS syncapi_current_room_state (
|
||||||
CONSTRAINT syncapi_room_state_unique UNIQUE (room_id, type, state_key)
|
CONSTRAINT syncapi_room_state_unique UNIQUE (room_id, type, state_key)
|
||||||
);
|
);
|
||||||
-- for event deletion
|
-- 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);
|
||||||
-- for querying membership states of users
|
-- 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';
|
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 = "" +
|
const upsertRoomStateSQL = "" +
|
||||||
"INSERT INTO syncapi_current_room_state (room_id, event_id, type, state_key, event_json, membership, added_at)" +
|
"INSERT INTO syncapi_current_room_state (room_id, event_id, type, sender, state_key, event_json, membership, added_at)" +
|
||||||
" VALUES ($1, $2, $3, $4, $5, $6, $7)" +
|
" VALUES ($1, $2, $3, $4, $5, $6, $7, 8)" +
|
||||||
" ON CONFLICT ON CONSTRAINT syncapi_room_state_unique" +
|
" 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, event_json = $6, membership = $7, added_at = $8"
|
||||||
|
|
||||||
const deleteRoomStateByEventIDSQL = "" +
|
const deleteRoomStateByEventIDSQL = "" +
|
||||||
"DELETE FROM syncapi_current_room_state WHERE event_id = $1"
|
"DELETE FROM syncapi_current_room_state WHERE event_id = $1"
|
||||||
|
|
@ -64,7 +68,12 @@ const selectRoomIDsWithMembershipSQL = "" +
|
||||||
"SELECT room_id FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = $1 AND membership = $2"
|
"SELECT room_id FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = $1 AND membership = $2"
|
||||||
|
|
||||||
const selectCurrentStateSQL = "" +
|
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 = ANY($4) )" +
|
||||||
|
" AND ( $5::text[] IS NULL OR NOT(type = ANY($5)) )" +
|
||||||
|
" LIMIT $6"
|
||||||
|
|
||||||
const selectJoinedUsersSQL = "" +
|
const selectJoinedUsersSQL = "" +
|
||||||
"SELECT room_id, state_key FROM syncapi_current_room_state WHERE type = 'm.room.member' AND membership = 'join'"
|
"SELECT room_id, state_key FROM syncapi_current_room_state WHERE type = 'm.room.member' AND membership = 'join'"
|
||||||
|
|
@ -165,10 +174,23 @@ func (s *currentRoomStateStatements) selectRoomIDsWithMembership(
|
||||||
|
|
||||||
// CurrentState returns all the current state events for the given room.
|
// CurrentState returns all the current state events for the given room.
|
||||||
func (s *currentRoomStateStatements) selectCurrentState(
|
func (s *currentRoomStateStatements) selectCurrentState(
|
||||||
ctx context.Context, txn *sql.Tx, roomID string,
|
ctx context.Context, txn *sql.Tx, roomID string, stateFilter *gomatrix.FilterPart,
|
||||||
) ([]gomatrixserverlib.Event, error) {
|
) ([]gomatrixserverlib.Event, error) {
|
||||||
|
|
||||||
|
var filter gomatrix.FilterPart
|
||||||
|
if stateFilter == nil {
|
||||||
|
filter = gomatrix.DefaultFilterPart()
|
||||||
|
} else {
|
||||||
|
filter = *stateFilter
|
||||||
|
}
|
||||||
|
|
||||||
stmt := common.TxStmt(txn, s.selectCurrentStateStmt)
|
stmt := common.TxStmt(txn, s.selectCurrentStateStmt)
|
||||||
rows, err := stmt.QueryContext(ctx, roomID)
|
rows, err := stmt.QueryContext(ctx, roomID,
|
||||||
|
pq.StringArray(filter.Senders),
|
||||||
|
pq.StringArray(filter.NotSenders),
|
||||||
|
pq.StringArray(filter.Types),
|
||||||
|
pq.StringArray(filter.NotTypes),
|
||||||
|
filter.Limit)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
@ -195,6 +217,7 @@ func (s *currentRoomStateStatements) upsertRoomState(
|
||||||
event.RoomID(),
|
event.RoomID(),
|
||||||
event.EventID(),
|
event.EventID(),
|
||||||
event.Type(),
|
event.Type(),
|
||||||
|
event.Sender(),
|
||||||
*event.StateKey(),
|
*event.StateKey(),
|
||||||
event.JSON(),
|
event.JSON(),
|
||||||
membership,
|
membership,
|
||||||
|
|
|
||||||
|
|
@ -178,10 +178,11 @@ func (d *SyncServerDatabase) GetStateEvent(
|
||||||
// Returns an empty slice if no state events could be found for this room.
|
// 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.
|
// Returns an error if there was an issue with the retrieval.
|
||||||
func (d *SyncServerDatabase) GetStateEventsForRoom(
|
func (d *SyncServerDatabase) GetStateEventsForRoom(
|
||||||
ctx context.Context, roomID string,
|
ctx context.Context, roomID string, stateFilter *gomatrix.FilterPart,
|
||||||
) (stateEvents []gomatrixserverlib.Event, err error) {
|
) (stateEvents []gomatrixserverlib.Event, err error) {
|
||||||
|
|
||||||
err = common.WithTransaction(d.db, func(txn *sql.Tx) 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, stateFilter)
|
||||||
return err
|
return err
|
||||||
})
|
})
|
||||||
return
|
return
|
||||||
|
|
@ -238,7 +239,7 @@ func (d *SyncServerDatabase) IncrementalSync(
|
||||||
// joined rooms, but also which rooms have membership transitions for this user between the 2 stream positions.
|
// 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
|
// This works out what the 'state' key should be for each room as well as which membership block
|
||||||
// to put the room into.
|
// 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, &filter.Room.State)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
@ -297,31 +298,37 @@ func (d *SyncServerDatabase) CompleteSync(
|
||||||
jr := types.NewJoinResponse()
|
jr := types.NewJoinResponse()
|
||||||
|
|
||||||
//Join response should contain events only if room isn't filtered
|
//Join response should contain events only if room isn't filtered
|
||||||
if !isRoomFiltered(roomID, filter, &filter.Room.Timeline) {
|
if !isRoomFiltered(roomID, filter, nil) {
|
||||||
// Timeline events
|
// Timeline events
|
||||||
var recentStreamEvents []streamEvent
|
var recentEvents []gomatrixserverlib.Event
|
||||||
var limited bool
|
if !isRoomFiltered(roomID, nil, &filter.Room.Timeline) {
|
||||||
recentStreamEvents, limited, err = d.events.selectRoomRecentEvents(
|
var recentStreamEvents []streamEvent
|
||||||
ctx, txn, roomID, posFrom, posTo, &filter.Room.Timeline)
|
var limited bool
|
||||||
if err != nil {
|
recentStreamEvents, limited, err = d.events.selectRoomRecentEvents(
|
||||||
return nil, err
|
ctx, txn, roomID, posFrom, posTo, &filter.Room.Timeline)
|
||||||
}
|
if err != nil {
|
||||||
jr.Timeline.Limited = limited
|
return nil, err
|
||||||
|
}
|
||||||
|
recentEvents = streamEventsToEvents(nil, recentStreamEvents)
|
||||||
|
|
||||||
recentEvents := streamEventsToEvents(nil, recentStreamEvents)
|
jr.Timeline.Limited = limited
|
||||||
jr.Timeline.Events = gomatrixserverlib.ToClientEvents(
|
jr.Timeline.Events = gomatrixserverlib.ToClientEvents(
|
||||||
recentEvents,
|
recentEvents,
|
||||||
gomatrixserverlib.FormatSync)
|
gomatrixserverlib.FormatSync)
|
||||||
|
}
|
||||||
|
|
||||||
// State events
|
// State events
|
||||||
var stateEvents []gomatrixserverlib.Event
|
if !isRoomFiltered(roomID, nil, &filter.Room.State) {
|
||||||
stateEvents, err = d.roomstate.selectCurrentState(ctx, txn, roomID)
|
var stateEvents []gomatrixserverlib.Event
|
||||||
if err != nil {
|
stateEvents, err = d.roomstate.selectCurrentState(ctx, txn, roomID, &filter.Room.State)
|
||||||
return nil, err
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if recentEvents != nil {
|
||||||
|
stateEvents = removeDuplicates(stateEvents, recentEvents)
|
||||||
|
}
|
||||||
|
jr.State.Events = gomatrixserverlib.ToClientEvents(stateEvents, gomatrixserverlib.FormatSync)
|
||||||
}
|
}
|
||||||
stateEvents = removeDuplicates(stateEvents, recentEvents)
|
|
||||||
jr.State.Events = gomatrixserverlib.ToClientEvents(stateEvents, gomatrixserverlib.FormatSync)
|
|
||||||
|
|
||||||
//TODO AccountData events
|
//TODO AccountData events
|
||||||
//TODO Ephemeral events
|
//TODO Ephemeral events
|
||||||
}
|
}
|
||||||
|
|
@ -564,7 +571,7 @@ func (d *SyncServerDatabase) fetchMissingStateEvents(
|
||||||
|
|
||||||
func (d *SyncServerDatabase) getStateDeltas(
|
func (d *SyncServerDatabase) getStateDeltas(
|
||||||
ctx context.Context, device *authtypes.Device, txn *sql.Tx,
|
ctx context.Context, device *authtypes.Device, txn *sql.Tx,
|
||||||
fromPos, toPos types.StreamPosition, userID string,
|
fromPos, toPos types.StreamPosition, userID string, stateFilter *gomatrix.FilterPart,
|
||||||
) ([]stateDelta, error) {
|
) ([]stateDelta, error) {
|
||||||
// Implement membership change algorithm: https://github.com/matrix-org/synapse/blob/v0.19.3/synapse/handlers/sync.py#L821
|
// 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
|
// - Get membership list changes for this user in this sync response
|
||||||
|
|
@ -597,7 +604,7 @@ func (d *SyncServerDatabase) getStateDeltas(
|
||||||
if membership == "join" {
|
if membership == "join" {
|
||||||
// send full room state down instead of a delta
|
// send full room state down instead of a delta
|
||||||
var allState []gomatrixserverlib.Event
|
var allState []gomatrixserverlib.Event
|
||||||
allState, err = d.roomstate.selectCurrentState(ctx, txn, roomID)
|
allState, err = d.roomstate.selectCurrentState(ctx, txn, roomID, stateFilter)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -55,22 +55,22 @@ func (filter *Filter) Validate() error {
|
||||||
|
|
||||||
func DefaultFilter() Filter {
|
func DefaultFilter() Filter {
|
||||||
return Filter{
|
return Filter{
|
||||||
AccountData: defaultFilterPart(),
|
AccountData: DefaultFilterPart(),
|
||||||
EventFields: nil,
|
EventFields: nil,
|
||||||
EventFormat: "client",
|
EventFormat: "client",
|
||||||
Presence: defaultFilterPart(),
|
Presence: DefaultFilterPart(),
|
||||||
Room: FilterRoom{
|
Room: FilterRoom{
|
||||||
AccountData: defaultFilterPart(),
|
AccountData: DefaultFilterPart(),
|
||||||
Ephemeral: defaultFilterPart(),
|
Ephemeral: DefaultFilterPart(),
|
||||||
IncludeLeave: false, //TODO check default value on synapse
|
IncludeLeave: false, //TODO check default value on synapse
|
||||||
NotRooms: nil,
|
NotRooms: nil,
|
||||||
Rooms: nil,
|
Rooms: nil,
|
||||||
State: defaultFilterPart(),
|
State: DefaultFilterPart(),
|
||||||
Timeline: defaultFilterPart(),
|
Timeline: DefaultFilterPart(),
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
func defaultFilterPart() FilterPart {
|
func DefaultFilterPart() FilterPart {
|
||||||
return FilterPart{
|
return FilterPart{
|
||||||
NotRooms: nil,
|
NotRooms: nil,
|
||||||
Rooms: nil,
|
Rooms: nil,
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue