feat: Collecting community messages count metrics (#3802)

* feat: proposal for collecting community metrics

https://github.com/status-im/status-desktop/issues/11152

* feat: collecting community message metrics with test

* feat: implement both strategies for fetching community metrics

* fix: review fixes

* fix: calc counts for timestamps

protocol/messenger_community_metrics.go

@@ -0,0 +1,113 @@
+package protocol
+
+import (
+	"errors"
+	"fmt"
+	"sort"
+
+	"github.com/status-im/status-go/eth-node/types"
+	"github.com/status-im/status-go/protocol/requests"
+)
+
+type MetricsIntervalResponse struct {
+	StartTimestamp uint64   `json:"startTimestamp"`
+	EndTimestamp   uint64   `json:"endTimestamp"`
+	Timestamps     []uint64 `json:"timestamps"`
+	Count          int      `json:"count"`
+}
+
+type CommunityMetricsResponse struct {
+	Type        requests.CommunityMetricsRequestType `json:"type"`
+	CommunityID types.HexBytes                       `json:"communityId"`
+	Intervals   []MetricsIntervalResponse            `json:"intervals"`
+}
+
+func (m *Messenger) getChatIdsForCommunity(communityID types.HexBytes) ([]string, error) {
+	community, err := m.GetCommunityByID(communityID)
+	if err != nil {
+		return []string{}, err
+	}
+
+	if community == nil {
+		return []string{}, errors.New("no community found")
+	}
+	return community.ChatIDs(), nil
+}
+
+func (m *Messenger) collectCommunityMessagesTimestamps(request *requests.CommunityMetricsRequest) (*CommunityMetricsResponse, error) {
+	chatIDs, err := m.getChatIdsForCommunity(request.CommunityID)
+	if err != nil {
+		return nil, err
+	}
+
+	intervals := make([]MetricsIntervalResponse, len(request.Intervals))
+	for i, sourceInterval := range request.Intervals {
+		// TODO: messages count should be stored in special table, not calculated here
+		timestamps, err := m.persistence.SelectMessagesTimestampsForChatsByPeriod(chatIDs, sourceInterval.StartTimestamp, sourceInterval.EndTimestamp)
+		if err != nil {
+			return nil, err
+		}
+
+		// there is no built-in sort for uint64
+		sort.Slice(timestamps, func(i, j int) bool { return timestamps[i] < timestamps[j] })
+
+		intervals[i] = MetricsIntervalResponse{
+			StartTimestamp: sourceInterval.StartTimestamp,
+			EndTimestamp:   sourceInterval.EndTimestamp,
+			Timestamps:     timestamps,
+			Count:          len(timestamps),
+		}
+	}
+
+	response := &CommunityMetricsResponse{
+		Type:        request.Type,
+		CommunityID: request.CommunityID,
+		Intervals:   intervals,
+	}
+
+	return response, nil
+}
+
+func (m *Messenger) collectCommunityMessagesCount(request *requests.CommunityMetricsRequest) (*CommunityMetricsResponse, error) {
+	chatIDs, err := m.getChatIdsForCommunity(request.CommunityID)
+	if err != nil {
+		return nil, err
+	}
+
+	intervals := make([]MetricsIntervalResponse, len(request.Intervals))
+	for i, sourceInterval := range request.Intervals {
+		// TODO: messages count should be stored in special table, not calculated here
+		count, err := m.persistence.SelectMessagesCountForChatsByPeriod(chatIDs, sourceInterval.StartTimestamp, sourceInterval.EndTimestamp)
+		if err != nil {
+			return nil, err
+		}
+		intervals[i] = MetricsIntervalResponse{
+			StartTimestamp: sourceInterval.StartTimestamp,
+			EndTimestamp:   sourceInterval.EndTimestamp,
+			Count:          count,
+		}
+	}
+
+	response := &CommunityMetricsResponse{
+		Type:        request.Type,
+		CommunityID: request.CommunityID,
+		Intervals:   intervals,
+	}
+
+	return response, nil
+}
+
+func (m *Messenger) CollectCommunityMetrics(request *requests.CommunityMetricsRequest) (*CommunityMetricsResponse, error) {
+	if err := request.Validate(); err != nil {
+		return nil, err
+	}
+
+	switch request.Type {
+	case requests.CommunityMetricsRequestMessagesTimestamps:
+		return m.collectCommunityMessagesTimestamps(request)
+	case requests.CommunityMetricsRequestMessagesCount:
+		return m.collectCommunityMessagesCount(request)
+	default:
+		return nil, fmt.Errorf("metrics for %d is not implemented yet", request.Type)
+	}
+}

protocol/messenger_community_metrics_test.go

@@ -0,0 +1,235 @@
+package protocol
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/status-im/status-go/eth-node/crypto"
+	"github.com/status-im/status-go/eth-node/types"
+	"github.com/status-im/status-go/protocol/common"
+	"github.com/status-im/status-go/protocol/communities"
+	"github.com/status-im/status-go/protocol/protobuf"
+	"github.com/status-im/status-go/protocol/requests"
+
+	"github.com/stretchr/testify/suite"
+)
+
+func TestMessengerCommunityMetricsSuite(t *testing.T) {
+	suite.Run(t, new(MessengerCommunityMetricsSuite))
+}
+
+type MessengerCommunityMetricsSuite struct {
+	MessengerBaseTestSuite
+}
+
+func (s *MessengerCommunityMetricsSuite) prepareCommunityAndChatIDs() (*communities.Community, []string) {
+	description := &requests.CreateCommunity{
+		Membership:  protobuf.CommunityPermissions_NO_MEMBERSHIP,
+		Name:        "status",
+		Color:       "#ffffff",
+		Description: "status community description",
+	}
+	response, err := s.m.CreateCommunity(description, true)
+	s.Require().NoError(err)
+	s.Require().NotNil(response)
+
+	s.Require().Len(response.Communities(), 1)
+	community := response.Communities()[0]
+
+	s.Require().Len(community.ChatIDs(), 1)
+	chatIDs := community.ChatIDs()
+
+	// Create another chat
+	chat := &protobuf.CommunityChat{
+		Permissions: &protobuf.CommunityPermissions{
+			Access: protobuf.CommunityPermissions_NO_MEMBERSHIP,
+		},
+		Identity: &protobuf.ChatIdentity{
+			DisplayName: "status",
+			Emoji:       "👍",
+			Description: "status community chat",
+		},
+	}
+	response, err = s.m.CreateCommunityChat(community.ID(), chat)
+	s.Require().NoError(err)
+	s.Require().NotNil(response)
+	s.Require().Len(response.Communities(), 1)
+	s.Require().Len(response.Chats(), 1)
+
+	chatIDs = append(chatIDs, response.Chats()[0].ID)
+
+	return community, chatIDs
+}
+
+func (s *MessengerCommunityMetricsSuite) prepareCommunityChatMessages(communityID string, chatIDs []string) {
+	s.generateMessages(chatIDs[0], communityID, []uint64{
+		// out ouf range messages in the beginning
+		1690162000,
+		// 1st column, 1 message
+		1690372200,
+		// 2nd column, 1 message
+		1690372800,
+		// 3rd column, 1 message
+		1690373000,
+		// out ouf range messages in the end
+		1690373100,
+	})
+
+	s.generateMessages(chatIDs[1], communityID, []uint64{
+		// out ouf range messages in the beginning
+		1690151000,
+		// 1st column, 2 messages
+		1690372000,
+		1690372100,
+		// 2nd column, 1 message
+		1690372700,
+		// 3rd column empty
+		// out ouf range messages in the end
+		1690373100,
+	})
+}
+
+func (s *MessengerCommunityMetricsSuite) generateMessages(chatID string, communityID string, timestamps []uint64) {
+	var messages []*common.Message
+	for i, timestamp := range timestamps {
+		message := &common.Message{
+			ChatMessage: protobuf.ChatMessage{
+				ChatId:      chatID,
+				Text:        fmt.Sprintf("Test message %d", i),
+				MessageType: protobuf.MessageType_ONE_TO_ONE,
+				// NOTE: should we filter content types for messages metrics
+				Clock:     timestamp,
+				Timestamp: timestamp,
+			},
+			WhisperTimestamp: timestamp,
+			From:             common.PubkeyToHex(&s.m.identity.PublicKey),
+			LocalChatID:      chatID,
+			CommunityID:      communityID,
+			ID:               types.EncodeHex(crypto.Keccak256([]byte(fmt.Sprintf("%s%s%d", chatID, communityID, timestamp)))),
+		}
+
+		err := message.PrepareContent(common.PubkeyToHex(&s.m.identity.PublicKey))
+		s.Require().NoError(err)
+
+		messages = append(messages, message)
+	}
+	err := s.m.persistence.SaveMessages(messages)
+	s.Require().NoError(err)
+}
+
+func (s *MessengerCommunityMetricsSuite) TestCollectCommunityMetricsInvalidRequest() {
+	community, _ := s.prepareCommunityAndChatIDs()
+
+	request := &requests.CommunityMetricsRequest{
+		CommunityID: community.ID(),
+		Type:        requests.CommunityMetricsRequestMessagesTimestamps,
+		Intervals: []requests.MetricsIntervalRequest{
+			requests.MetricsIntervalRequest{
+				StartTimestamp: 1690372400,
+				EndTimestamp:   1690371800,
+			},
+			requests.MetricsIntervalRequest{
+				StartTimestamp: 1690371900,
+				EndTimestamp:   1690373000,
+			},
+		},
+	}
+
+	// Expect error
+	_, err := s.m.CollectCommunityMetrics(request)
+	s.Require().Error(err)
+	s.Require().Equal(err, requests.ErrInvalidTimestampIntervals)
+}
+
+func (s *MessengerCommunityMetricsSuite) TestCollectCommunityMetricsEmptyInterval() {
+	community, _ := s.prepareCommunityAndChatIDs()
+
+	request := &requests.CommunityMetricsRequest{
+		CommunityID: community.ID(),
+		Type:        requests.CommunityMetricsRequestMessagesTimestamps,
+	}
+
+	// Expect empty metrics
+	resp, err := s.m.CollectCommunityMetrics(request)
+	s.Require().NoError(err)
+	s.Require().NotNil(resp)
+
+	// Entries count should be empty
+	s.Require().Len(resp.Intervals, 0)
+}
+
+func (s *MessengerCommunityMetricsSuite) TestCollectCommunityMessagesTimestamps() {
+	community, chatIDs := s.prepareCommunityAndChatIDs()
+
+	s.prepareCommunityChatMessages(string(community.ID()), chatIDs)
+
+	// Request metrics
+	request := &requests.CommunityMetricsRequest{
+		CommunityID: community.ID(),
+		Type:        requests.CommunityMetricsRequestMessagesTimestamps,
+		Intervals: []requests.MetricsIntervalRequest{
+			requests.MetricsIntervalRequest{
+				StartTimestamp: 1690372000,
+				EndTimestamp:   1690372300,
+			},
+			requests.MetricsIntervalRequest{
+				StartTimestamp: 1690372400,
+				EndTimestamp:   1690372800,
+			},
+			requests.MetricsIntervalRequest{
+				StartTimestamp: 1690372900,
+				EndTimestamp:   1690373000,
+			},
+		},
+	}
+
+	resp, err := s.m.CollectCommunityMetrics(request)
+	s.Require().NoError(err)
+	s.Require().NotNil(resp)
+
+	s.Require().Len(resp.Intervals, 3)
+
+	s.Require().Equal(resp.Intervals[0].Count, 3)
+	s.Require().Equal(resp.Intervals[1].Count, 2)
+	s.Require().Equal(resp.Intervals[2].Count, 1)
+
+	s.Require().Equal(resp.Intervals[0].Timestamps, []uint64{1690372000, 1690372100, 1690372200})
+	s.Require().Equal(resp.Intervals[1].Timestamps, []uint64{1690372700, 1690372800})
+	s.Require().Equal(resp.Intervals[2].Timestamps, []uint64{1690373000})
+}
+
+func (s *MessengerCommunityMetricsSuite) TestCollectCommunityMessagesCount() {
+	community, chatIDs := s.prepareCommunityAndChatIDs()
+
+	s.prepareCommunityChatMessages(string(community.ID()), chatIDs)
+
+	// Request metrics
+	request := &requests.CommunityMetricsRequest{
+		CommunityID: community.ID(),
+		Type:        requests.CommunityMetricsRequestMessagesCount,
+		Intervals: []requests.MetricsIntervalRequest{
+			requests.MetricsIntervalRequest{
+				StartTimestamp: 1690372000,
+				EndTimestamp:   1690372300,
+			},
+			requests.MetricsIntervalRequest{
+				StartTimestamp: 1690372400,
+				EndTimestamp:   1690372800,
+			},
+			requests.MetricsIntervalRequest{
+				StartTimestamp: 1690372900,
+				EndTimestamp:   1690373000,
+			},
+		},
+	}
+
+	resp, err := s.m.CollectCommunityMetrics(request)
+	s.Require().NoError(err)
+	s.Require().NotNil(resp)
+
+	s.Require().Len(resp.Intervals, 3)
+
+	s.Require().Equal(resp.Intervals[0].Count, 3)
+	s.Require().Equal(resp.Intervals[1].Count, 2)
+	s.Require().Equal(resp.Intervals[2].Count, 1)
+}

protocol/persistence_metrics.go

@@ -0,0 +1,63 @@
+package protocol
+
+import (
+	"database/sql"
+	"fmt"
+	"strings"
+)
+
+const selectTimestampsQuery = "SELECT whisper_timestamp FROM user_messages WHERE %s whisper_timestamp >= ? AND whisper_timestamp <= ?"
+const selectCountQuery = "SELECT COUNT(*) FROM user_messages WHERE %s whisper_timestamp >= ? AND whisper_timestamp <= ?"
+
+func querySeveralChats(chatIDs []string) string {
+	if len(chatIDs) == 0 {
+		return ""
+	}
+
+	var conditions []string
+	for _, chatID := range chatIDs {
+		conditions = append(conditions, fmt.Sprintf("local_chat_id = '%s'", chatID))
+	}
+	return fmt.Sprintf("(%s) AND", strings.Join(conditions, " OR "))
+}
+
+func (db sqlitePersistence) SelectMessagesTimestampsForChatsByPeriod(chatIDs []string, startTimestamp uint64, endTimestamp uint64) ([]uint64, error) {
+	query := fmt.Sprintf(selectTimestampsQuery, querySeveralChats(chatIDs))
+
+	rows, err := db.db.Query(query, startTimestamp, endTimestamp)
+	if err != nil {
+		return []uint64{}, err
+	}
+	defer rows.Close()
+
+	var timestamps []uint64
+	for rows.Next() {
+		var timestamp uint64
+		err := rows.Scan(&timestamp)
+		if err != nil {
+			return nil, err
+		}
+		timestamps = append(timestamps, timestamp)
+	}
+
+	err = rows.Err()
+	if err != nil {
+		return []uint64{}, err
+	}
+
+	return timestamps, nil
+}
+
+func (db sqlitePersistence) SelectMessagesCountForChatsByPeriod(chatIDs []string, startTimestamp uint64, endTimestamp uint64) (int, error) {
+	query := fmt.Sprintf(selectCountQuery, querySeveralChats(chatIDs))
+
+	var count int
+	if err := db.db.QueryRow(query, startTimestamp, endTimestamp).Scan(&count); err != nil {
+		if err == sql.ErrNoRows {
+			return 0, nil
+		}
+		return 0, err
+	}
+
+	return count, nil
+}

protocol/requests/community_metrics_request.go

@@ -0,0 +1,44 @@
+package requests
+
+import (
+	"errors"
+
+	"github.com/status-im/status-go/eth-node/types"
+)
+
+var ErrNoCommunityID = errors.New("community metrics request has no community id")
+var ErrInvalidTimestampIntervals = errors.New("community metrics request invalid time intervals")
+
+type CommunityMetricsRequestType uint
+
+const (
+	CommunityMetricsRequestMessagesTimestamps CommunityMetricsRequestType = iota
+	CommunityMetricsRequestMessagesCount
+	CommunityMetricsRequestMembers
+	CommunityMetricsRequestControlNodeUptime
+)
+
+type MetricsIntervalRequest struct {
+	StartTimestamp uint64 `json:"startTimestamp"`
+	EndTimestamp   uint64 `json:"endTimestamp"`
+}
+
+type CommunityMetricsRequest struct {
+	CommunityID types.HexBytes              `json:"communityId"`
+	Type        CommunityMetricsRequestType `json:"type"`
+	Intervals   []MetricsIntervalRequest    `json:"intervals"`
+}
+
+func (r *CommunityMetricsRequest) Validate() error {
+	if len(r.CommunityID) == 0 {
+		return ErrNoCommunityID
+	}
+
+	for _, interval := range r.Intervals {
+		if interval.StartTimestamp == 0 || interval.EndTimestamp == 0 || interval.StartTimestamp >= interval.EndTimestamp {
+			return ErrInvalidTimestampIntervals
+		}
+	}
+
+	return nil
+}

services/ext/api.go

@@ -1368,6 +1368,10 @@ func (api *PublicAPI) CheckAllCommunityChannelsPermissions(request *requests.Che
 	return api.service.messenger.CheckAllCommunityChannelsPermissions(request)
 }
 
+func (api *PublicAPI) CollectCommunityMetrics(request *requests.CommunityMetricsRequest) (*protocol.CommunityMetricsResponse, error) {
+	return api.service.messenger.CollectCommunityMetrics(request)
+}
+
 func (api *PublicAPI) ShareCommunityURLWithChatKey(communityID types.HexBytes) (string, error) {
 	return api.service.messenger.ShareCommunityURLWithChatKey(communityID)
 }