Limit the amount of chat messages

Author: xaionaro

cmd/streamcli/commands/commands.go

@@ -319,7 +319,7 @@ func chatListen(cmd *cobra.Command, args []string) {
 	assertNoError(ctx, err)
 
 	fmt.Println("subscribing...")
-	ch, err := streamD.SubscribeToChatMessages(ctx, time.Now().Add(-time.Minute))
+	ch, err := streamD.SubscribeToChatMessages(ctx, time.Now().Add(-time.Minute), 1000)
 	assertNoError(ctx, err)
 
 	fmt.Println("started listening...")

cmd/streampanel/FyneApp.toml

@@ -5,4 +5,4 @@ Website = "https://github.com/xaionaro/streamctl"
   Name = "streampanel"
   ID = "center.dx.streampanel"
   Version = "0.1.0"
-  Build = 411
+  Build = 412

pkg/chatmessagesstorage/get_messages_since.go

@@ -14,16 +14,20 @@ import (
 func (s *ChatMessagesStorage) GetMessagesSince(
 	ctx context.Context,
 	since time.Time,
+	limit uint,
 ) ([]api.ChatMessage, error) {
-	return xsync.DoA2R2(ctx, &s.Mutex, s.getMessagesSinceLocked, ctx, since)
+	return xsync.DoA3R2(ctx, &s.Mutex, s.getMessagesSinceLocked, ctx, since, limit)
 }
 
 func (s *ChatMessagesStorage) getMessagesSinceLocked(
 	ctx context.Context,
 	since time.Time,
+	limit uint,
 ) (_ret []api.ChatMessage, _err error) {
-	logger.Tracef(ctx, "getMessagesSinceLocked(ctx, %v)", since)
-	defer func() { logger.Tracef(ctx, "/getMessagesSinceLocked(ctx, %v): len:%d, %v", since, len(_ret), _err) }()
+	logger.Tracef(ctx, "getMessagesSinceLocked(ctx, %v, %d)", since, limit)
+	defer func() {
+		logger.Tracef(ctx, "/getMessagesSinceLocked(ctx, %v, %d): len:%d, %v", since, limit, len(_ret), _err)
+	}()
 
 	if len(s.Messages) == 0 {
 		return nil, nil
@@ -46,5 +50,9 @@ func (s *ChatMessagesStorage) getMessagesSinceLocked(
 		idx = 0
 	}
 
+	if limit > 0 && len(s.Messages)-idx > int(limit) {
+		idx = len(s.Messages) - int(limit)
+	}
+
 	return slices.Clone(s.Messages[idx:]), nil
 }

pkg/streamd/api/streamd.go

@@ -279,6 +279,7 @@ type StreamD interface {
 	SubscribeToChatMessages(
 		ctx context.Context,
 		since time.Time,
+		limit uint64,
 	) (<-chan ChatMessage, error)
 	SendChatMessage(
 		ctx context.Context,

pkg/streamd/chat.go

@@ -16,7 +16,7 @@ type ChatMessageStorage interface {
 	RemoveMessage(context.Context, streamcontrol.ChatMessageID) error
 	Load(ctx context.Context) error
 	Store(ctx context.Context) error
-	GetMessagesSince(context.Context, time.Time) ([]api.ChatMessage, error)
+	GetMessagesSince(context.Context, time.Time, uint) ([]api.ChatMessage, error)
 }
 
 func (d *StreamD) startListeningForChatMessages(
@@ -102,11 +102,12 @@ func (d *StreamD) BanUser(
 func (d *StreamD) SubscribeToChatMessages(
 	ctx context.Context,
 	since time.Time,
+	limit uint64,
 ) (<-chan api.ChatMessage, error) {
-	return eventSubToChan[api.ChatMessage](
+	return eventSubToChan(
 		ctx, d,
 		func(ctx context.Context, outCh chan api.ChatMessage) {
-			msgs, err := d.ChatMessagesStorage.GetMessagesSince(ctx, since)
+			msgs, err := d.ChatMessagesStorage.GetMessagesSince(ctx, since, uint(limit))
 			if err != nil {
 				logger.Errorf(ctx, "unable to get the messages from the storage: %v", err)
 				return

pkg/streamd/client/client.go

@@ -2631,6 +2631,7 @@ func (c *Client) SubmitEvent(
 func (c *Client) SubscribeToChatMessages(
 	ctx context.Context,
 	since time.Time,
+	limit uint64,
 ) (<-chan api.ChatMessage, error) {
 	return unwrapStreamDChan(
 		ctx,
@@ -2645,6 +2646,7 @@ func (c *Client) SubscribeToChatMessages(
 				client.SubscribeToChatMessages,
 				&streamd_grpc.SubscribeToChatMessagesRequest{
 					SinceUNIXNano: uint64(since.UnixNano()),
+					Limit:         limit,
 				},
 			)
 		},

pkg/streamd/grpc/go/streamd_grpc/streamd.pb.go

@@ -747,6 +747,7 @@ message SubmitEventReply {}
 
 message SubscribeToChatMessagesRequest {
     uint64 sinceUNIXNano = 1;
+    uint64 limit = 2;
 }
 message ChatMessage {
 	uint64 createdAtUNIXNano  = 1;

pkg/streamd/grpc/streamd.proto

@@ -1913,13 +1913,14 @@ func (grpc *GRPCServer) SubscribeToChatMessages(
 	srv streamd_grpc.StreamD_SubscribeToChatMessagesServer,
 ) error {
 	ts := req.GetSinceUNIXNano()
+	limit := req.GetLimit()
 	since := time.Unix(
 		int64(ts)/int64(time.Second.Nanoseconds()),
 		int64(ts)%int64(time.Second.Nanoseconds()),
 	)
 	return wrapChan(
 		func(ctx context.Context) (<-chan api.ChatMessage, error) {
-			return grpc.StreamD.SubscribeToChatMessages(ctx, since)
+			return grpc.StreamD.SubscribeToChatMessages(ctx, since, limit)
 		},
 		srv,
 		func(input api.ChatMessage) streamd_grpc.ChatMessage {

pkg/streamd/server/grpc.go

@@ -17,7 +17,7 @@ import (
 )
 
 const (
-	ChatLogSize = 100
+	ChatLogSize = 50
 )
 
 type chatUIInterface interface {
@@ -66,7 +66,7 @@ func (p *Panel) getChatUIs(ctx context.Context) []chatUIInterface {
 }
 
 func (p *Panel) initChatMessagesHandler(ctx context.Context) error {
-	msgCh, err := p.StreamD.SubscribeToChatMessages(ctx, time.Now().Add(-7*24*time.Hour))
+	msgCh, err := p.StreamD.SubscribeToChatMessages(ctx, time.Now().Add(-7*24*time.Hour), ChatLogSize)
 	if err != nil {
 		return fmt.Errorf("unable to subscribe to chat messages: %w", err)
 	}