Fix invitations not sending push notifications (#3207)

The tests added in https://github.com/matrix-org/sytest/pull/1356
uncovered that we don't consider invitations as events the userapi
should handle and thus just don't notify the client about any new
invitations received over federation.

sytest-whitelist

@@ -784,3 +784,7 @@ Invited user can reject local invite after originator leaves
 Guest users can join guest_access rooms
 Forgotten room messages cannot be paginated
 Local device key changes get to remote servers with correct prev_id
+HS provides query metadata
+HS can provide query metadata on a single protocol
+Invites over federation are correctly pushed
+Invites over federation are correctly pushed with name

userapi/consumers/roomserver.go

@@ -92,18 +92,36 @@ 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 {
+
-		return true
+	var event *rstypes.HeaderedEvent
-	}
+	var isNewRoomEvent bool
-	var output rsapi.OutputEvent
+	switch rsapi.OutputType(msg.Header.Get(jetstream.RoomEventType)) {
-	if err := json.Unmarshal(msg.Data, &output); err != nil {
+	case rsapi.OutputTypeNewRoomEvent:
-		// If the message was invalid, log it and move on to the next message in the stream
+		isNewRoomEvent = true
-		log.WithError(err).Errorf("roomserver output log: message parse failure")
+		fallthrough
-		return true
+	case rsapi.OutputTypeNewInviteEvent:
-	}
+		var output rsapi.OutputEvent
-	event := output.NewRoomEvent.Event
+		if err := json.Unmarshal(msg.Data, &output); err != nil {
-	if event == nil {
+			// If the message was invalid, log it and move on to the next message in the stream
-		log.Errorf("userapi consumer: expected event")
+			log.WithError(err).Errorf("roomserver output log: message parse failure")
+			return true
+		}
+		if isNewRoomEvent {
+			event = output.NewRoomEvent.Event
+		} else {
+			event = output.NewInviteEvent.Event
+		}
+
+		if event == nil {
+			log.Errorf("userapi consumer: expected event")
+			return true
+		}
+
+		log.WithFields(log.Fields{
+			"event_id":   event.EventID(),
+			"event_type": event.Type(),
+		}).Tracef("Received message from roomserver: %#v", output)
+	default:
 		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,19 @@ 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 &&
+		gjson.GetBytes(event.Content(), "membership").Str == "invite" {
+		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},