feat: update EventEmitter with EventBus from orbitdb

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

go.mod

@@ -54,6 +54,7 @@ require (
 	github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99
 	github.com/juju/fslock v0.0.0-20160525022230-4d5c94c67b4b
 	github.com/kr/pretty v0.3.0
+	github.com/libp2p/go-eventbus v0.2.1
 	github.com/libp2p/go-libp2p v0.17.0
 	github.com/libp2p/go-libp2p-core v0.13.0
 	github.com/libp2p/go-libp2p-discovery v0.6.0
@@ -98,6 +99,7 @@ require (
 	golang.org/x/crypto v0.0.0-20210921155107-089bfa567519
 	golang.org/x/mobile v0.0.0-20210220033013-bdb1ca9a1e08
 	golang.org/x/net v0.0.0-20210813160813-60bc85c4be6d
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 	golang.org/x/text v0.3.7
 	golang.org/x/tools v0.1.7
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1

go/cmd/berty/mini/view_group.go

@@ -358,6 +358,7 @@ func (v *groupView) loop(ctx context.Context) {
 					}
 
 					// @TODO: Log this
+
 					v.syncMessages <- &historyMessage{
 						messageType: messageTypeError,
 						payload:     []byte(err.Error()),

go/internal/cryptoutil/keystore_message.go

@@ -381,42 +381,64 @@ func (m *MessageKeystore) putKeyForCID(ctx context.Context, id cid.Cid, key *[32
 	return nil
 }
 
-func (m *MessageKeystore) OpenEnvelope(ctx context.Context, g *protocoltypes.Group, ownPK crypto.PubKey, data []byte, id cid.Cid) (*protocoltypes.MessageHeaders, *protocoltypes.EncryptedMessage, [][]byte, error) {
-	if m == nil || g == nil {
-		return nil, nil, nil, errcode.ErrInvalidInput
-	}
-
+func (m *MessageKeystore) OpenEnvelopePayload(
+	ctx context.Context,
+	env *protocoltypes.MessageEnvelope,
+	headers *protocoltypes.MessageHeaders,
+	g *protocoltypes.Group,
+	ownPK crypto.PubKey,
+	id cid.Cid,
+) (*protocoltypes.EncryptedMessage, [][]byte, error) {
 	gPK, err := g.GetPubKey()
 	if err != nil {
-		return nil, nil, nil, errcode.ErrDeserialization.Wrap(err)
-	}
-
-	env, headers, err := OpenEnvelopeHeaders(data, g)
-	if err != nil {
-		return nil, nil, nil, errcode.ErrCryptoDecrypt.Wrap(err)
+		return nil, nil, errcode.ErrDeserialization.Wrap(err)
 	}
 
 	msgBytes, decryptInfo, err := m.OpenPayload(ctx, id, gPK, env.Message, headers)
 	if err != nil {
-		return headers, nil, nil, errcode.ErrCryptoDecryptPayload.Wrap(err)
+		return nil, nil, errcode.ErrCryptoDecryptPayload.Wrap(err)
 	}
 
 	if err := m.PostDecryptActions(ctx, decryptInfo, g, ownPK, headers); err != nil {
-		return nil, nil, nil, errcode.TODO.Wrap(err)
+		return nil, nil, errcode.TODO.Wrap(err)
 	}
 
 	var msg protocoltypes.EncryptedMessage
 	err = msg.Unmarshal(msgBytes)
 	if err != nil {
-		return nil, nil, nil, errcode.ErrDeserialization.Wrap(err)
+		return nil, nil, errcode.ErrDeserialization.Wrap(err)
 	}
 
 	attachmentsCIDs, err := AttachmentCIDSliceDecrypt(g, env.GetEncryptedAttachmentCIDs())
+	if err != nil {
+		return nil, nil, errcode.ErrCryptoDecrypt.Wrap(err)
+	}
+
+	return &msg, attachmentsCIDs, nil
+}
+
+func (m *MessageKeystore) OpenEnvelope(
+	ctx context.Context,
+	g *protocoltypes.Group,
+	ownPK crypto.PubKey,
+	data []byte, id cid.Cid,
+) (*protocoltypes.MessageHeaders, *protocoltypes.EncryptedMessage, [][]byte, error) {
+	if m == nil || g == nil {
+		return nil, nil, nil, errcode.ErrInvalidInput
+	}
+
+	env, headers, err := OpenEnvelopeHeaders(data, g)
 	if err != nil {
 		return nil, nil, nil, errcode.ErrCryptoDecrypt.Wrap(err)
 	}
 
-	return headers, &msg, attachmentsCIDs, nil
+	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
 }
 
 func (m *MessageKeystore) OpenPayload(ctx context.Context, id cid.Cid, groupPK crypto.PubKey, payload []byte, headers *protocoltypes.MessageHeaders) ([]byte, *DecryptInfo, error) {

go/internal/ipfsutil/testing.go

@@ -176,10 +176,7 @@ func TestingCoreAPIUsingMockNet(ctx context.Context, t testing.TB, opts *Testing
 		}
 
 		ps, err = pubsub.NewGossipSub(ctx, h,
-			pubsub.WithMessageSigning(true),
-			pubsub.WithFloodPublish(true),
 			pubsub.WithDiscovery(disc),
-			pubsub.WithPeerExchange(true),
 			pubsubtracker.EventTracerOption(),
 		)
 

go/pkg/bertyprotocol/api_debug.go

@@ -84,7 +84,7 @@ func (s *service) DebugInspectGroupStore(req *protocoltypes.DebugInspectGroupSto
 				nexts    = make([][]byte, len(e.GetNext()))
 			)
 
-			if evt, err := cg.messageStore.openMessage(srv.Context(), e, false); err != nil {
+			if evt, err := cg.messageStore.openMessage(srv.Context(), e); err != nil {
 				s.logger.Error("unable to open message", zap.Error(err))
 			} else {
 				devicePK = evt.Headers.DevicePK