1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
|
package streams
import (
"context"
"crypto/sha256"
"encoding/base64"
"math"
"strconv"
"time"
"github.com/matrix-org/gomatrixserverlib/spec"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/syncapi/storage"
"github.com/matrix-org/dendrite/syncapi/synctypes"
"github.com/matrix-org/dendrite/syncapi/types"
)
type InviteStreamProvider struct {
DefaultStreamProvider
rsAPI api.SyncRoomserverAPI
}
func (p *InviteStreamProvider) Setup(
ctx context.Context, snapshot storage.DatabaseTransaction,
) {
p.DefaultStreamProvider.Setup(ctx, snapshot)
p.latestMutex.Lock()
defer p.latestMutex.Unlock()
id, err := snapshot.MaxStreamPositionForInvites(ctx)
if err != nil {
panic(err)
}
p.latest = id
}
func (p *InviteStreamProvider) CompleteSync(
ctx context.Context,
snapshot storage.DatabaseTransaction,
req *types.SyncRequest,
) types.StreamPosition {
return p.IncrementalSync(ctx, snapshot, req, 0, p.LatestPosition(ctx))
}
func (p *InviteStreamProvider) IncrementalSync(
ctx context.Context,
snapshot storage.DatabaseTransaction,
req *types.SyncRequest,
from, to types.StreamPosition,
) types.StreamPosition {
r := types.Range{
From: from,
To: to,
}
invites, retiredInvites, maxID, err := snapshot.InviteEventsInRange(
ctx, req.Device.UserID, r,
)
if err != nil {
req.Log.WithError(err).Error("p.DB.InviteEventsInRange failed")
return from
}
eventFormat := synctypes.FormatSync
if req.Filter.EventFormat == synctypes.EventFormatFederation {
eventFormat = synctypes.FormatSyncFederation
}
for roomID, inviteEvent := range invites {
user := spec.UserID{}
sender, err := p.rsAPI.QueryUserIDForSender(ctx, inviteEvent.RoomID(), inviteEvent.SenderID())
if err == nil && sender != nil {
user = *sender
}
sk := inviteEvent.StateKey()
if sk != nil && *sk != "" {
skUserID, err := p.rsAPI.QueryUserIDForSender(ctx, inviteEvent.RoomID(), spec.SenderID(*inviteEvent.StateKey()))
if err == nil && skUserID != nil {
skString := skUserID.String()
sk = &skString
}
}
// skip ignored user events
if _, ok := req.IgnoredUsers.List[user.String()]; ok {
continue
}
ir := types.NewInviteResponse(inviteEvent, user, sk, eventFormat)
req.Response.Rooms.Invite[roomID] = ir
}
// When doing an initial sync, we don't want to add retired invites, as this
// can add rooms we were invited to, but already left.
if from == 0 {
return to
}
for roomID := range retiredInvites {
membership, _, err := snapshot.SelectMembershipForUser(ctx, roomID, req.Device.UserID, math.MaxInt64)
// Skip if the user is an existing member of the room.
// Otherwise, the NewLeaveResponse will eject the user from the room unintentionally
if membership == spec.Join ||
err != nil {
continue
}
lr := types.NewLeaveResponse()
h := sha256.Sum256(append([]byte(roomID), []byte(strconv.FormatInt(int64(to), 10))...))
lr.Timeline.Events = append(lr.Timeline.Events, synctypes.ClientEvent{
// fake event ID which muxes in the to position
EventID: "$" + base64.RawURLEncoding.EncodeToString(h[:]),
OriginServerTS: spec.AsTimestamp(time.Now()),
RoomID: roomID,
Sender: req.Device.UserID,
StateKey: &req.Device.UserID,
Type: "m.room.member",
Content: spec.RawJSON(`{"membership":"leave"}`),
})
req.Response.Rooms.Leave[roomID] = lr
}
return maxID
}
|
