⚡ Optimize for passive users

✨ Subscribe to channel focused

pkg/internal/grpc/stream.go

@@ -3,6 +3,8 @@ package grpc
 import (
 	"context"
 	"fmt"
+	"strings"
+
 	"git.solsynth.dev/hypernet/messaging/pkg/internal/gap"
 	"git.solsynth.dev/hypernet/messaging/pkg/internal/http/exts"
 	"git.solsynth.dev/hypernet/messaging/pkg/internal/services"
@@ -51,6 +53,35 @@ func (v *Server) PushStream(_ context.Context, request *proto.PushStreamRequest)
 			})
 			break
 		}
+	case "events.subscribe", "events.unsubscribe", "events.unsubscribeAll":
+		var data struct {
+			ChannelID uint `json:"channel_id" validate:"required"`
+		}
+
+		err := jsoniter.Unmarshal(in.RawPayload(), &data)
+		if err == nil {
+			err = exts.ValidateStruct(data)
+		}
+		if err != nil {
+			_, _ = sc.PushStream(context.Background(), &proto.PushStreamRequest{
+				ClientId: request.ClientId,
+				Body: nex.WebSocketPackage{
+					Action:  "error",
+					Message: fmt.Sprintf("unable parse payload: %v", err),
+				}.Marshal(),
+			})
+			break
+		}
+
+		action := strings.Split(in.Action, ".")[1]
+		switch action {
+		case "subscribe":
+			services.SubscribeChannel(uint(request.GetUserId()), data.ChannelID)
+		case "unsubscribe":
+			services.UnsubscribeChannel(uint(request.GetUserId()), data.ChannelID)
+		case "unsubscribeAll":
+			services.UnsubscribeAll(uint(request.GetUserId()))
+		}
 	case "events.read":
 		var data struct {
 			ChannelMemberID uint `json:"channel_member_id" validate:"required"`

pkg/internal/models/channels.go

@@ -39,7 +39,7 @@ func (v Channel) DisplayText() string {
 	if v.Realm != nil {
 		return fmt.Sprintf("%s, %s", v.Name, v.Realm.Name)
 	}
-	return fmt.Sprintf("%s", v.Name)
+	return v.Name
 }
 
 type NotifyLevel = int8

pkg/internal/services/calls.go

@@ -117,10 +117,6 @@ func NewCall(channel models.Channel, founder models.ChannelMember) (models.Call,
 			if member.ID != call.Founder.ID {
 				pendingUsers = append(pendingUsers, uint64(member.AccountID))
 			}
-			PushCommand(member.AccountID, nex.WebSocketPackage{
-				Action:  "calls.new",
-				Payload: call,
-			})
 		}
 
 		channel, _ = GetChannel(channel.ID)
@@ -131,6 +127,13 @@ func NewCall(channel models.Channel, founder models.ChannelMember) (models.Call,
 			}
 		}
 
+		// The call notification is not happen very often
+		// So we don't need to optimize the performance for passive users
+		PushCommandBatch(pendingUsers, nex.WebSocketPackage{
+			Action:  "calls.new",
+			Payload: call,
+		})
+
 		err = authkit.NotifyUserBatch(
 			gap.Nx,
 			pendingUsers,
@@ -172,12 +175,17 @@ func EndCall(call models.Call) (models.Call, error) {
 		ChannelID: call.ChannelID,
 	}).Find(&members).Error; err == nil {
 		call, _ = GetCall(call.Channel, call.ID)
+		var pendingUsers []uint64
 		for _, member := range members {
-			PushCommand(member.AccountID, nex.WebSocketPackage{
-				Action:  "calls.end",
-				Payload: call,
-			})
+			if member.ID != call.Founder.ID {
+				pendingUsers = append(pendingUsers, uint64(member.AccountID))
+			}
 		}
+
+		PushCommandBatch(pendingUsers, nex.WebSocketPackage{
+			Action:  "calls.end",
+			Payload: call,
+		})
 	}
 
 	return call, nil

pkg/internal/services/channels.go

@@ -297,6 +297,8 @@ func EditChannel(channel models.Channel) (models.Channel, error) {
 
 func DeleteChannel(channel models.Channel) error {
 	if err := database.C.Delete(&channel).Error; err == nil {
+		UnsubscribeAllWithChannels(channel.ID)
+
 		database.C.Where("channel_id = ?", channel.ID).Delete(&models.Event{})
 
 		cacheManager := cache.New[any](localCache.S)

pkg/internal/services/events.go

@@ -15,6 +15,7 @@ import (
 	jsoniter "github.com/json-iterator/go"
 	"github.com/rs/zerolog/log"
 	"github.com/samber/lo"
+	"github.com/spf13/viper"
 )
 
 func CountEvent(channel models.Channel) int64 {
@@ -89,7 +90,13 @@ func NewEvent(event models.Event) (models.Event, error) {
 		log.Error().Err(err).Msg("Failed to fetch event, the notifying of new event was terminated...")
 		return event, err
 	}
-	idxList := lo.Map(members, func(item models.ChannelMember, index int) uint64 {
+	idxList := lo.Map(lo.Filter(members, func(item models.ChannelMember, index int) bool {
+		if !viper.GetBool("performance.passive_user_optimize") {
+			// Leave this for backward compatibility
+			return true
+		}
+		return CheckSubscribed(item.AccountID, event.ChannelID)
+	}), func(item models.ChannelMember, index int) uint64 {
 		return uint64(item.AccountID)
 	})
 	_ = PushCommandBatch(idxList, nex.WebSocketPackage{
@@ -120,6 +127,9 @@ func NotifyMessageEvent(members []models.ChannelMember, event models.Event) {
 	var mentionedUsers []uint64
 
 	for _, member := range members {
+		if CheckSubscribed(member.AccountID, event.ChannelID) {
+			continue
+		}
 		if member.ID != event.SenderID {
 			switch member.Notify {
 			case models.NotifyLevelNone:

pkg/internal/services/status.go

@@ -3,15 +3,16 @@ package services
 import (
 	"context"
 	"fmt"
+
 	localCache "git.solsynth.dev/hypernet/messaging/pkg/internal/cache"
 	"git.solsynth.dev/hypernet/messaging/pkg/internal/database"
-	"git.solsynth.dev/hypernet/messaging/pkg/internal/gap"
 	"git.solsynth.dev/hypernet/messaging/pkg/internal/models"
 	"git.solsynth.dev/hypernet/nexus/pkg/nex"
-	"git.solsynth.dev/hypernet/nexus/pkg/proto"
 	"github.com/eko/gocache/lib/v4/cache"
 	"github.com/eko/gocache/lib/v4/marshaler"
 	"github.com/eko/gocache/lib/v4/store"
+	"github.com/samber/lo"
+	"github.com/spf13/viper"
 )
 
 type statusQueryCacheEntry struct {
@@ -76,14 +77,18 @@ func SetTypingStatus(channelId uint, userId uint) error {
 		)
 	}
 
-	sc := proto.NewStreamServiceClient(gap.Nx.GetNexusGrpcConn())
-	_, err := sc.PushStreamBatch(context.Background(), &proto.PushStreamBatchRequest{
-		UserId: broadcastTarget,
-		Body: nex.WebSocketPackage{
-			Action:  "status.typing",
-			Payload: data,
-		}.Marshal(),
+	broadcastTarget = lo.Filter(broadcastTarget, func(item uint64, index int) bool {
+		if !viper.GetBool("performance.passive_user_optimize") {
+			// Leave this for backward compatibility
+			return true
+		}
+		return CheckSubscribed(uint(item), channelId)
 	})
 
-	return err
+	PushCommandBatch(broadcastTarget, nex.WebSocketPackage{
+		Action:  "status.typing",
+		Payload: data,
+	})
+
+	return nil
 }

pkg/internal/services/subscribe.go

@@ -0,0 +1,51 @@
+package services
+
+import "sync"
+
+// ChannelID -> UserID
+var subscribeInfo = make(map[uint]map[uint]bool)
+var subscribeLock sync.Mutex
+
+// If user subscribed to a channel
+// Push the new message to them via websocket
+// And skip the notification
+
+func CheckSubscribed(UserID uint, ChannelID uint) bool {
+	if _, ok := subscribeInfo[ChannelID]; ok {
+		if _, ok := subscribeInfo[ChannelID][UserID]; ok {
+			return true
+		}
+	}
+	return false
+}
+
+func SubscribeChannel(userId uint, channelId uint) {
+	subscribeLock.Lock()
+	defer subscribeLock.Unlock()
+	if _, ok := subscribeInfo[channelId]; !ok {
+		subscribeInfo[channelId] = make(map[uint]bool)
+	}
+	subscribeInfo[channelId][userId] = true
+}
+
+func UnsubscribeChannel(userId uint, channelId uint) {
+	subscribeLock.Lock()
+	defer subscribeLock.Unlock()
+	if _, ok := subscribeInfo[channelId]; ok {
+		delete(subscribeInfo[channelId], userId)
+	}
+}
+
+func UnsubscribeAll(userId uint) {
+	subscribeLock.Lock()
+	defer subscribeLock.Unlock()
+	for _, v := range subscribeInfo {
+		delete(v, userId)
+	}
+}
+
+func UnsubscribeAllWithChannels(channelId uint) {
+	subscribeLock.Lock()
+	defer subscribeLock.Unlock()
+	delete(subscribeInfo, channelId)
+}

settings.toml

@@ -5,6 +5,9 @@ grpc_bind = "0.0.0.0:7447"
 
 nexus_addr = "localhost:7001"
 
+[performance]
+passive_user_optimize = true
+
 [debug]
 database = false
 print_routes = false