mirror of
https://github.com/1f349/dendrite.git
synced 2024-11-09 22:42:58 +00:00
Send presence to joined hosts only (#2858)
Send presence events only to rooms the user is participating, not all servers we know about. Should fix #2752
This commit is contained in:
parent
eeabe892a9
commit
b13cb43785
@ -22,6 +22,7 @@ import (
|
|||||||
"github.com/matrix-org/dendrite/federationapi/queue"
|
"github.com/matrix-org/dendrite/federationapi/queue"
|
||||||
"github.com/matrix-org/dendrite/federationapi/storage"
|
"github.com/matrix-org/dendrite/federationapi/storage"
|
||||||
fedTypes "github.com/matrix-org/dendrite/federationapi/types"
|
fedTypes "github.com/matrix-org/dendrite/federationapi/types"
|
||||||
|
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
|
||||||
"github.com/matrix-org/dendrite/setup/config"
|
"github.com/matrix-org/dendrite/setup/config"
|
||||||
"github.com/matrix-org/dendrite/setup/jetstream"
|
"github.com/matrix-org/dendrite/setup/jetstream"
|
||||||
"github.com/matrix-org/dendrite/setup/process"
|
"github.com/matrix-org/dendrite/setup/process"
|
||||||
@ -39,6 +40,7 @@ type OutputPresenceConsumer struct {
|
|||||||
db storage.Database
|
db storage.Database
|
||||||
queues *queue.OutgoingQueues
|
queues *queue.OutgoingQueues
|
||||||
isLocalServerName func(gomatrixserverlib.ServerName) bool
|
isLocalServerName func(gomatrixserverlib.ServerName) bool
|
||||||
|
rsAPI roomserverAPI.FederationRoomserverAPI
|
||||||
topic string
|
topic string
|
||||||
outboundPresenceEnabled bool
|
outboundPresenceEnabled bool
|
||||||
}
|
}
|
||||||
@ -50,6 +52,7 @@ func NewOutputPresenceConsumer(
|
|||||||
js nats.JetStreamContext,
|
js nats.JetStreamContext,
|
||||||
queues *queue.OutgoingQueues,
|
queues *queue.OutgoingQueues,
|
||||||
store storage.Database,
|
store storage.Database,
|
||||||
|
rsAPI roomserverAPI.FederationRoomserverAPI,
|
||||||
) *OutputPresenceConsumer {
|
) *OutputPresenceConsumer {
|
||||||
return &OutputPresenceConsumer{
|
return &OutputPresenceConsumer{
|
||||||
ctx: process.Context(),
|
ctx: process.Context(),
|
||||||
@ -60,6 +63,7 @@ func NewOutputPresenceConsumer(
|
|||||||
durable: cfg.Matrix.JetStream.Durable("FederationAPIPresenceConsumer"),
|
durable: cfg.Matrix.JetStream.Durable("FederationAPIPresenceConsumer"),
|
||||||
topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputPresenceEvent),
|
topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputPresenceEvent),
|
||||||
outboundPresenceEnabled: cfg.Matrix.Presence.EnableOutbound,
|
outboundPresenceEnabled: cfg.Matrix.Presence.EnableOutbound,
|
||||||
|
rsAPI: rsAPI,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -89,6 +93,16 @@ func (t *OutputPresenceConsumer) onMessage(ctx context.Context, msgs []*nats.Msg
|
|||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var queryRes roomserverAPI.QueryRoomsForUserResponse
|
||||||
|
err = t.rsAPI.QueryRoomsForUser(t.ctx, &roomserverAPI.QueryRoomsForUserRequest{
|
||||||
|
UserID: userID,
|
||||||
|
WantMembership: "join",
|
||||||
|
}, &queryRes)
|
||||||
|
if err != nil {
|
||||||
|
log.WithError(err).Error("failed to calculate joined rooms for user")
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
presence := msg.Header.Get("presence")
|
presence := msg.Header.Get("presence")
|
||||||
|
|
||||||
ts, err := strconv.Atoi(msg.Header.Get("last_active_ts"))
|
ts, err := strconv.Atoi(msg.Header.Get("last_active_ts"))
|
||||||
@ -96,11 +110,13 @@ func (t *OutputPresenceConsumer) onMessage(ctx context.Context, msgs []*nats.Msg
|
|||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
joined, err := t.db.GetAllJoinedHosts(ctx)
|
// send this presence to all servers who share rooms with this user.
|
||||||
|
joined, err := t.db.GetJoinedHostsForRooms(t.ctx, queryRes.RoomIDs, true)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.WithError(err).Error("failed to get joined hosts")
|
log.WithError(err).Error("failed to get joined hosts")
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
if len(joined) == 0 {
|
if len(joined) == 0 {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
@ -164,7 +164,7 @@ func NewInternalAPI(
|
|||||||
}
|
}
|
||||||
|
|
||||||
presenceConsumer := consumers.NewOutputPresenceConsumer(
|
presenceConsumer := consumers.NewOutputPresenceConsumer(
|
||||||
base.ProcessContext, cfg, js, queues, federationDB,
|
base.ProcessContext, cfg, js, queues, federationDB, rsAPI,
|
||||||
)
|
)
|
||||||
if err = presenceConsumer.Start(); err != nil {
|
if err = presenceConsumer.Start(); err != nil {
|
||||||
logrus.WithError(err).Panic("failed to start presence consumer")
|
logrus.WithError(err).Panic("failed to start presence consumer")
|
||||||
|
@ -46,3 +46,6 @@ If a device list update goes missing, the server resyncs on the next one
|
|||||||
# Might be a bug in the test because leaves do appear :-(
|
# Might be a bug in the test because leaves do appear :-(
|
||||||
|
|
||||||
Leaves are present in non-gapped incremental syncs
|
Leaves are present in non-gapped incremental syncs
|
||||||
|
|
||||||
|
# Below test was passing for the wrong reason, failing correctly since #2858
|
||||||
|
New federated private chats get full presence information (SYN-115)
|
@ -682,7 +682,6 @@ Presence changes are reported to local room members
|
|||||||
Presence changes are also reported to remote room members
|
Presence changes are also reported to remote room members
|
||||||
Presence changes to UNAVAILABLE are reported to local room members
|
Presence changes to UNAVAILABLE are reported to local room members
|
||||||
Presence changes to UNAVAILABLE are reported to remote room members
|
Presence changes to UNAVAILABLE are reported to remote room members
|
||||||
New federated private chats get full presence information (SYN-115)
|
|
||||||
/upgrade copies >100 power levels to the new room
|
/upgrade copies >100 power levels to the new room
|
||||||
Room state after a rejected message event is the same as before
|
Room state after a rejected message event is the same as before
|
||||||
Room state after a rejected state event is the same as before
|
Room state after a rejected state event is the same as before
|
||||||
|
Loading…
Reference in New Issue
Block a user