2022-04-06 13:11:19 +02:00
|
|
|
// Copyright 2022 The Matrix.org Foundation C.I.C.
|
|
|
|
//
|
|
|
|
// 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 streams
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"encoding/json"
|
2022-12-08 08:25:03 +01:00
|
|
|
"fmt"
|
2022-04-06 13:11:19 +02:00
|
|
|
"sync"
|
|
|
|
|
2022-08-19 15:32:24 +02:00
|
|
|
"github.com/tidwall/gjson"
|
|
|
|
|
2022-04-06 13:11:19 +02:00
|
|
|
"github.com/matrix-org/dendrite/syncapi/notifier"
|
2022-09-30 13:48:10 +02:00
|
|
|
"github.com/matrix-org/dendrite/syncapi/storage"
|
2023-04-04 19:16:53 +02:00
|
|
|
"github.com/matrix-org/dendrite/syncapi/synctypes"
|
2022-04-06 13:11:19 +02:00
|
|
|
"github.com/matrix-org/dendrite/syncapi/types"
|
2023-04-19 16:50:33 +02:00
|
|
|
"github.com/matrix-org/gomatrixserverlib/spec"
|
2022-04-06 13:11:19 +02:00
|
|
|
)
|
|
|
|
|
|
|
|
type PresenceStreamProvider struct {
|
2022-09-30 13:48:10 +02:00
|
|
|
DefaultStreamProvider
|
2022-04-06 13:11:19 +02:00
|
|
|
// cache contains previously sent presence updates to avoid unneeded updates
|
|
|
|
cache sync.Map
|
|
|
|
notifier *notifier.Notifier
|
|
|
|
}
|
|
|
|
|
2022-09-30 13:48:10 +02:00
|
|
|
func (p *PresenceStreamProvider) Setup(
|
|
|
|
ctx context.Context, snapshot storage.DatabaseTransaction,
|
|
|
|
) {
|
|
|
|
p.DefaultStreamProvider.Setup(ctx, snapshot)
|
2022-04-06 13:11:19 +02:00
|
|
|
|
2022-09-30 13:48:10 +02:00
|
|
|
p.latestMutex.Lock()
|
|
|
|
defer p.latestMutex.Unlock()
|
|
|
|
|
|
|
|
id, err := snapshot.MaxStreamPositionForPresence(ctx)
|
2022-04-06 13:11:19 +02:00
|
|
|
if err != nil {
|
|
|
|
panic(err)
|
|
|
|
}
|
|
|
|
p.latest = id
|
|
|
|
}
|
|
|
|
|
|
|
|
func (p *PresenceStreamProvider) CompleteSync(
|
|
|
|
ctx context.Context,
|
2022-09-30 13:48:10 +02:00
|
|
|
snapshot storage.DatabaseTransaction,
|
2022-04-06 13:11:19 +02:00
|
|
|
req *types.SyncRequest,
|
|
|
|
) types.StreamPosition {
|
2022-09-30 13:48:10 +02:00
|
|
|
return p.IncrementalSync(ctx, snapshot, req, 0, p.LatestPosition(ctx))
|
2022-04-06 13:11:19 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
func (p *PresenceStreamProvider) IncrementalSync(
|
|
|
|
ctx context.Context,
|
2022-09-30 13:48:10 +02:00
|
|
|
snapshot storage.DatabaseTransaction,
|
2022-04-06 13:11:19 +02:00
|
|
|
req *types.SyncRequest,
|
|
|
|
from, to types.StreamPosition,
|
|
|
|
) types.StreamPosition {
|
2022-04-28 16:12:40 +02:00
|
|
|
// We pull out a larger number than the filter asks for, since we're filtering out events later
|
2023-04-04 19:16:53 +02:00
|
|
|
presences, err := snapshot.PresenceAfter(ctx, from, synctypes.EventFilter{Limit: 1000})
|
2022-04-06 13:11:19 +02:00
|
|
|
if err != nil {
|
|
|
|
req.Log.WithError(err).Error("p.DB.PresenceAfter failed")
|
|
|
|
return from
|
|
|
|
}
|
|
|
|
|
2022-12-08 08:25:03 +01:00
|
|
|
getPresenceForUsers, err := p.getNeededUsersFromRequest(ctx, req, presences)
|
|
|
|
if err != nil {
|
|
|
|
req.Log.WithError(err).Error("getNeededUsersFromRequest failed")
|
|
|
|
return from
|
|
|
|
}
|
|
|
|
|
|
|
|
// Got no presence between range and no presence to get from the database
|
|
|
|
if len(getPresenceForUsers) == 0 && len(presences) == 0 {
|
2022-04-06 13:11:19 +02:00
|
|
|
return to
|
|
|
|
}
|
|
|
|
|
2022-12-08 08:25:03 +01:00
|
|
|
dbPresences, err := snapshot.GetPresences(ctx, getPresenceForUsers)
|
|
|
|
if err != nil {
|
|
|
|
req.Log.WithError(err).Error("unable to query presence for user")
|
|
|
|
_ = snapshot.Rollback()
|
|
|
|
return from
|
|
|
|
}
|
|
|
|
for _, presence := range dbPresences {
|
|
|
|
presences[presence.UserID] = presence
|
2022-04-06 13:11:19 +02:00
|
|
|
}
|
|
|
|
|
2022-04-28 16:12:40 +02:00
|
|
|
lastPos := from
|
2022-04-27 12:25:07 +02:00
|
|
|
for _, presence := range presences {
|
|
|
|
if presence == nil {
|
|
|
|
continue
|
|
|
|
}
|
2022-04-06 13:11:19 +02:00
|
|
|
// Ignore users we don't share a room with
|
2022-04-06 14:31:44 +02:00
|
|
|
if req.Device.UserID != presence.UserID && !p.notifier.IsSharedUser(req.Device.UserID, presence.UserID) {
|
2022-04-06 13:11:19 +02:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
cacheKey := req.Device.UserID + req.Device.ID + presence.UserID
|
|
|
|
pres, ok := p.cache.Load(cacheKey)
|
|
|
|
if ok {
|
|
|
|
// skip already sent presence
|
|
|
|
prevPresence := pres.(*types.PresenceInternal)
|
|
|
|
currentlyActive := prevPresence.CurrentlyActive()
|
|
|
|
skip := prevPresence.Equals(presence) && currentlyActive && req.Device.UserID != presence.UserID
|
2022-10-19 14:05:39 +02:00
|
|
|
_, membershipChange := req.MembershipChanges[presence.UserID]
|
|
|
|
if skip && !membershipChange {
|
2022-06-30 12:34:37 +02:00
|
|
|
req.Log.Tracef("Skipping presence, no change (%s)", presence.UserID)
|
2022-04-06 13:11:19 +02:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
}
|
2022-04-07 11:10:28 +02:00
|
|
|
|
|
|
|
if _, known := types.PresenceFromString(presence.ClientFields.Presence); known {
|
|
|
|
presence.ClientFields.LastActiveAgo = presence.LastActiveAgo()
|
|
|
|
if presence.ClientFields.Presence == "online" {
|
|
|
|
currentlyActive := presence.CurrentlyActive()
|
|
|
|
presence.ClientFields.CurrentlyActive = ¤tlyActive
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
presence.ClientFields.Presence = "offline"
|
2022-04-06 13:11:19 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
content, err := json.Marshal(presence.ClientFields)
|
|
|
|
if err != nil {
|
|
|
|
return from
|
|
|
|
}
|
|
|
|
|
2023-04-04 19:16:53 +02:00
|
|
|
req.Response.Presence.Events = append(req.Response.Presence.Events, synctypes.ClientEvent{
|
2022-04-06 13:11:19 +02:00
|
|
|
Content: content,
|
|
|
|
Sender: presence.UserID,
|
2023-04-19 16:50:33 +02:00
|
|
|
Type: spec.MPresence,
|
2022-04-06 13:11:19 +02:00
|
|
|
})
|
|
|
|
if presence.StreamPos > lastPos {
|
|
|
|
lastPos = presence.StreamPos
|
|
|
|
}
|
2022-04-28 16:12:40 +02:00
|
|
|
if len(req.Response.Presence.Events) == req.Filter.Presence.Limit {
|
|
|
|
break
|
|
|
|
}
|
2022-04-06 13:11:19 +02:00
|
|
|
p.cache.Store(cacheKey, presence)
|
|
|
|
}
|
|
|
|
|
2022-04-30 00:07:50 +02:00
|
|
|
if len(req.Response.Presence.Events) == 0 {
|
|
|
|
return to
|
|
|
|
}
|
|
|
|
|
2022-04-06 13:11:19 +02:00
|
|
|
return lastPos
|
|
|
|
}
|
|
|
|
|
2022-12-08 08:25:03 +01:00
|
|
|
func (p *PresenceStreamProvider) getNeededUsersFromRequest(ctx context.Context, req *types.SyncRequest, presences map[string]*types.PresenceInternal) ([]string, error) {
|
|
|
|
getPresenceForUsers := []string{}
|
|
|
|
// Add presence for users which newly joined a room
|
|
|
|
for userID := range req.MembershipChanges {
|
|
|
|
if _, ok := presences[userID]; ok {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
getPresenceForUsers = append(getPresenceForUsers, userID)
|
|
|
|
}
|
|
|
|
|
|
|
|
// add newly joined rooms user presences
|
|
|
|
newlyJoined := joinedRooms(req.Response, req.Device.UserID)
|
|
|
|
if len(newlyJoined) == 0 {
|
|
|
|
return getPresenceForUsers, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// TODO: Check if this is working better than before.
|
|
|
|
if err := p.notifier.LoadRooms(ctx, p.DB, newlyJoined); err != nil {
|
|
|
|
return getPresenceForUsers, fmt.Errorf("unable to refresh notifier lists: %w", err)
|
|
|
|
}
|
|
|
|
for _, roomID := range newlyJoined {
|
|
|
|
roomUsers := p.notifier.JoinedUsers(roomID)
|
|
|
|
for i := range roomUsers {
|
|
|
|
// we already got a presence from this user
|
|
|
|
if _, ok := presences[roomUsers[i]]; ok {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
getPresenceForUsers = append(getPresenceForUsers, roomUsers[i])
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return getPresenceForUsers, nil
|
|
|
|
}
|
|
|
|
|
2022-04-06 13:11:19 +02:00
|
|
|
func joinedRooms(res *types.Response, userID string) []string {
|
|
|
|
var roomIDs []string
|
|
|
|
for roomID, join := range res.Rooms.Join {
|
|
|
|
// we would expect to see our join event somewhere if we newly joined the room.
|
|
|
|
// Normal events get put in the join section so it's not enough to know the room ID is present in 'join'.
|
|
|
|
newlyJoined := membershipEventPresent(join.State.Events, userID)
|
|
|
|
if newlyJoined {
|
|
|
|
roomIDs = append(roomIDs, roomID)
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
newlyJoined = membershipEventPresent(join.Timeline.Events, userID)
|
|
|
|
if newlyJoined {
|
|
|
|
roomIDs = append(roomIDs, roomID)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return roomIDs
|
|
|
|
}
|
|
|
|
|
2023-04-04 19:16:53 +02:00
|
|
|
func membershipEventPresent(events []synctypes.ClientEvent, userID string) bool {
|
2022-04-06 13:11:19 +02:00
|
|
|
for _, ev := range events {
|
|
|
|
// it's enough to know that we have our member event here, don't need to check membership content
|
|
|
|
// as it's implied by being in the respective section of the sync response.
|
2023-04-19 16:50:33 +02:00
|
|
|
if ev.Type == spec.MRoomMember && ev.StateKey != nil && *ev.StateKey == userID {
|
2022-08-19 15:32:24 +02:00
|
|
|
// ignore e.g. join -> join changes
|
|
|
|
if gjson.GetBytes(ev.Unsigned, "prev_content.membership").Str == gjson.GetBytes(ev.Content, "membership").Str {
|
|
|
|
continue
|
|
|
|
}
|
2022-04-06 13:11:19 +02:00
|
|
|
return true
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return false
|
|
|
|
}
|