feat: optimize store message cache

Signed-off-by: gfanton <8671905+gfanton@users.noreply.github.com>

go/cmd/berty/mini/view_group.go

@@ -353,19 +353,16 @@ func (v *groupView) loop(ctx context.Context) {
 			for {
 				evt, err = cl.Recv()
 				if err != nil {
-					if err == io.EOF {
-						return
+					if err != io.EOF {
+						v.syncMessages <- &historyMessage{
+							messageType: messageTypeError,
+							payload:     []byte(err.Error()),
+						}
 					}
-
-					// @TODO: Log this
-
-					v.syncMessages <- &historyMessage{
-						messageType: messageTypeError,
-						payload:     []byte(err.Error()),
-					}
-					continue
+					return
 				}
 
+				// @TODO: Log this
 				metadataEventHandler(ctx, v, evt, false, v.logger)
 			}
 		}()

go/internal/cryptoutil/keystore_message.go

@@ -62,6 +62,16 @@ func (m *MessageKeystore) GetDeviceChainKey(ctx context.Context, groupPK, pk cry
 	return ds, nil
 }
 
+func (m *MessageKeystore) HasSecretForRawDevicePK(ctx context.Context, groupPK, devicePK []byte) (has bool) {
+	if m == nil {
+		return false
+	}
+
+	key := idForCurrentCK(groupPK, devicePK)
+	has, _ = m.store.Has(ctx, key)
+	return
+}
+
 func (m *MessageKeystore) delPrecomputedKey(ctx context.Context, groupPK, device crypto.PubKey, counter uint64) error {
 	if m == nil {
 		return errcode.ErrInvalidInput
@@ -435,7 +445,6 @@ func (m *MessageKeystore) OpenEnvelope(
 	msg, attachmentsCIDs, err := m.OpenEnvelopePayload(ctx, env, headers, g, ownPK, id)
 	if err != nil {
 		return nil, nil, nil, errcode.TODO.Wrap(err)
-
 	}
 
 	return headers, msg, attachmentsCIDs, nil

go/internal/handshake/request.go

@@ -131,6 +131,7 @@ func (hc *handshakeContext) receiveResponderAccept() error {
 	if err != nil {
 		return errcode.ErrHandshakeResponderAcceptBoxKeyGen.Wrap(err)
 	}
+
 	respBytes, _ := box.OpenAfterPrecomputation(
 		nil,
 		boxEnvelope.Box,

go/internal/ipfsutil/repo.go

@@ -15,7 +15,6 @@ import (
 	"github.com/pkg/errors"
 
 	"berty.tech/berty/v2/go/pkg/errcode"
-	// nolint:staticcheck
 	encrepo "berty.tech/go-ipfs-repo-encrypted"
 )
 

go/pkg/bertyprotocol/api_event.go

@@ -6,12 +6,9 @@ import (
 	"fmt"
 
 	"github.com/libp2p/go-eventbus"
-	"go.uber.org/zap"
 
 	"berty.tech/berty/v2/go/pkg/errcode"
 	"berty.tech/berty/v2/go/pkg/protocoltypes"
-
-	"berty.tech/go-orbit-db/stores"
 )
 
 func checkParametersConsistency(sinceID, untilID []byte, sinceNow, untilNow, reverseOrder bool) error {
@@ -103,8 +100,6 @@ func (s *service) GroupMetadataList(req *protocoltypes.GroupMetadataList_Request
 	}
 
 	// Subscribe to new metadata events and stream them if requested
-	listPreviouseMetadataDone := false
-	bufferMetadata := []*protocoltypes.GroupMetadataEvent{}
 	for {
 		var event interface{}
 		select {
@@ -114,43 +109,16 @@ func (s *service) GroupMetadataList(req *protocoltypes.GroupMetadataList_Request
 		case event = <-newEvents:
 		}
 
-		var metadatas []*protocoltypes.GroupMetadataEvent
-		switch evt := event.(type) {
-		case stores.EventReplicated:
-			entries := evt.Entries
-			metadatas = []*protocoltypes.GroupMetadataEvent{}
-			cg.logger.Info("receving replicated metadata events", zap.Int("metadatas", len(entries)))
-			for _, entry := range entries {
-				msg, _, err := cg.MetadataStore().openMetadataEntry(entry)
-				if err != nil {
-					s.logger.Error("unable to open metadata", zap.Error(err))
-					continue
-				}
-
-				metadatas = append(metadatas, msg)
-			}
-
-			if !listPreviouseMetadataDone {
-				bufferMetadata = append(bufferMetadata, metadatas...)
-				continue
-			}
-
-		case protocoltypes.GroupMetadataEvent:
-			if evt.EventContext == nil {
-				listPreviouseMetadataDone = true
-				metadatas = bufferMetadata
-			} else {
-				metadatas = []*protocoltypes.GroupMetadataEvent{&evt}
-			}
+		msg := event.(protocoltypes.GroupMetadataEvent)
+		if msg.EventContext == nil {
+			continue
 		}
 
-		for _, msg := range metadatas {
-			if err := sub.Send(msg); err != nil {
-				return err
-			}
+		if err := sub.Send(&msg); err != nil {
+			return err
 		}
 
-		cg.logger.Info("service - metadata store - sent event from log subscription", zap.Int("metadatas", len(metadatas)))
+		cg.logger.Info("service - metadata store - sent 1 event from log subscription")
 	}
 }
 
@@ -174,7 +142,6 @@ func (s *service) GroupMessageList(req *protocoltypes.GroupMessageList_Request,
 	var newEvents <-chan interface{}
 	if req.UntilID == nil && !req.UntilNow {
 		sub, err := cg.MessageStore().EventBus().Subscribe([]interface{}{
-			new(stores.EventReplicated),
 			new(protocoltypes.GroupMessageEvent),
 		}, eventbus.BufSize(32))
 		if err != nil {
@@ -223,7 +190,6 @@ func (s *service) GroupMessageList(req *protocoltypes.GroupMessageList_Request,
 
 	// Subscribe to new message events and stream them if requested
 	// listPreviouseMessageDone := false
-	bufferMessage := []*protocoltypes.GroupMessageEvent{}
 	for {
 		var event interface{}
 		select {
@@ -233,46 +199,15 @@ func (s *service) GroupMessageList(req *protocoltypes.GroupMessageList_Request,
 		case event = <-newEvents:
 		}
 
-		var messages []*protocoltypes.GroupMessageEvent
-		switch evt := event.(type) {
-		case stores.EventReplicated:
-			entries := evt.Entries
-			messages = []*protocoltypes.GroupMessageEvent{}
-			for _, entry := range entries {
-				if err := cg.MessageStore().addToMessageQueue(ctx, entry); err != nil {
-					s.logger.Error("unable to add message to queue", zap.Error(err))
-				}
-				// msg, err := cg.MessageStore().openMessage(ctx, entry, false)
-				// if err != nil {
-				// 	s.logger.Error("unable to open message", zap.Error(err))
-				// 	continue
-				// } else {
-				// 	messages = append(messages, msg)
-				// }
-			}
-
+		msg := event.(protocoltypes.GroupMessageEvent)
+		if msg.EventContext == nil {
 			continue
-
-			// if !listPreviouseMessageDone {
-			// 	bufferMessage = append(bufferMessage, messages...)
-			// 	continue
-			// }
-
-		case protocoltypes.GroupMessageEvent:
-			if evt.EventContext == nil {
-				// listPreviouseMessageDone = true
-				messages = bufferMessage
-			} else {
-				messages = []*protocoltypes.GroupMessageEvent{&evt}
-			}
 		}
 
-		for _, msg := range messages {
-			if err := sub.Send(msg); err != nil {
-				return err
-			}
+		if err := sub.Send(&msg); err != nil {
+			return err
 		}
 
-		cg.logger.Info("service - message store - sent event from log subscription", zap.Int("messages", len(messages)))
+		cg.logger.Info("service - message store - sent 1 event from log subscription")
 	}
 }

go/pkg/bertyprotocol/group.go

@@ -151,7 +151,7 @@ func FillMessageKeysHolderUsingNewData(ctx context.Context, gc *GroupContext) <-
 
 			// A new chainKey is registered, check if cached messages can be opened with it
 			if rawPK, err := pk.Raw(); err == nil {
-				gc.MessageStore().ProcessMessageQueueForDevicePK(ctx, rawPK)
+				gc.MessageStore().ProcessMessageQueueForDevicePK(rawPK)
 			}
 
 			ch <- pk