Method onMessage

userapi/consumers/syncapi_readupdate.go:67–136

(ctx context.Context, msg *nats.Msg)

}

func (s *OutputReadUpdateConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
	var read types.ReadUpdate
	if err := json.Unmarshal(msg.Data, &read); err != nil {
		log.WithError(err).Error("userapi clientapi consumer: message parse failure")
		return true
	}
	if read.FullyRead == 0 && read.Read == 0 {
		return true
	}

	userID := string(msg.Header.Get(jetstream.UserID))
	roomID := string(msg.Header.Get(jetstream.RoomID))

	localpart, domain, err := gomatrixserverlib.SplitID('@', userID)
	if err != nil {
		log.WithError(err).Error("userapi clientapi consumer: SplitID failure")
		return true
	}
	if domain != s.ServerName {
		log.Error("userapi clientapi consumer: not a local user")
		return true
	}

	log := log.WithFields(log.Fields{
		"room_id": roomID,
		"user_id": userID,
	})
	log.Tracef("Received read update from sync API: %#v", read)

	if read.Read > 0 {
		updated, err := s.db.SetNotificationsRead(ctx, localpart, roomID, int64(read.Read), true)
		if err != nil {
			log.WithError(err).Error("userapi EDU consumer")
			return false
		}

		if updated {
			if err = s.syncProducer.GetAndSendNotificationData(ctx, userID, roomID); err != nil {
				log.WithError(err).Error("userapi EDU consumer: GetAndSendNotificationData failed")
				return false
			}
			if err = util.NotifyUserCountsAsync(ctx, s.pgClient, localpart, s.db); err != nil {
				log.WithError(err).Error("userapi EDU consumer: NotifyUserCounts failed")
				return false
			}
		}
	}

	if read.FullyRead > 0 {
		deleted, err := s.db.DeleteNotificationsUpTo(ctx, localpart, roomID, int64(read.FullyRead))
		if err != nil {
			log.WithError(err).Errorf("userapi clientapi consumer: DeleteNotificationsUpTo failed")
			return false
		}

		if deleted {
			if err := util.NotifyUserCountsAsync(ctx, s.pgClient, localpart, s.db); err != nil {
				log.WithError(err).Error("userapi clientapi consumer: NotifyUserCounts failed")