mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-12-07 06:53:09 -06:00
Code Issues Packages Projects Releases Wiki Activity

Fix notifications on invites

Browse source
This commit is contained in:
Till Faelligen 2023-09-26 14:08:32 +02:00
parent db83789654
commit b6affa2e37
No known key found for this signature in database
GPG key ID: 3DF82D8AB9211D4E
1 changed files with 38 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

51
userapi/consumers/roomserver.go
View file

@ -92,16 +92,34 @@ func (s *OutputRoomEventConsumer) Start() error {
func (s *OutputRoomEventConsumer) onMessage(ctx context.Context, msgs []*nats.Msg) bool {
msg := msgs[0] // Guaranteed to exist if onMessage is called
// Only handle events we care about
if rsapi.OutputType(msg.Header.Get(jetstream.RoomEventType)) != rsapi.OutputTypeNewRoomEvent {
var event *rstypes.HeaderedEvent
var isNewRoomEvent bool
switch rsapi.OutputType(msg.Header.Get(jetstream.RoomEventType)) {
case rsapi.OutputTypeNewRoomEvent:
isNewRoomEvent = true
fallthrough
case rsapi.OutputTypeNewInviteEvent:
var output rsapi.OutputEvent
if err := json.Unmarshal(msg.Data, &output); err != nil {
// If the message was invalid, log it and move on to the next message in the stream
log.WithError(err).Errorf("roomserver output log: message parse failure")
return true
}
if isNewRoomEvent {
event = output.NewRoomEvent.Event
} else {
event = output.NewInviteEvent.Event
}
log.WithFields(log.Fields{
"event_id": event.EventID(),
"event_type": event.Type(),
}).Tracef("Received message from roomserver: %#v", output)
default:
return true
}
var output rsapi.OutputEvent
if err := json.Unmarshal(msg.Data, &output); err != nil {
// If the message was invalid, log it and move on to the next message in the stream
log.WithError(err).Errorf("roomserver output log: message parse failure")
return true
}
event := output.NewRoomEvent.Event
if event == nil {
log.Errorf("userapi consumer: expected event")
return true
@ -111,11 +129,6 @@ func (s *OutputRoomEventConsumer) onMessage(ctx context.Context, msgs []*nats.Ms
go s.storeMessageStats(ctx, event.Type(), string(event.SenderID()), event.RoomID().String())
}
log.WithFields(log.Fields{
"event_id": event.EventID(),
"event_type": event.Type(),
}).Tracef("Received message from roomserver: %#v", output)
metadata, err := msg.Metadata()
if err != nil {
return true
@ -448,6 +461,18 @@ func (s *OutputRoomEventConsumer) roomName(ctx context.Context, event *rstypes.H
}
}
// Special case for invites, as we don't store any "current state" for these events,
// we need to make sure that, if present, the m.room.name is sent as well.
if event.Type() == spec.MRoomMember {
invState := gjson.GetBytes(event.JSON(), "unsigned.invite_room_state")
for _, ev := range invState.Array() {
if ev.Get("type").Str == spec.MRoomName {
name := ev.Get("content.name").Str
return name, nil
}
}
}
req := &rsapi.QueryCurrentStateRequest{
RoomID: event.RoomID().String(),
StateTuples: []gomatrixserverlib.StateKeyTuple{roomNameTuple, canonicalAliasTuple},