refactor: use protected topics persistence interface

messaging/adapters/persistence.go

@@ -1,8 +1,11 @@
 package adapters
 
 import (
+	"crypto/ecdsa"
+
 	"github.com/status-im/status-go/messaging/layers/transport"
 	"github.com/status-im/status-go/messaging/types"
+	wakupersistence "github.com/status-im/status-go/messaging/waku/persistence"
 )
 
 type KeysPersistence struct {
@@ -37,3 +40,36 @@ func (pm *ProcessedMessageIDsCache) Add(ids []string, timestamp uint64) error {
 func (pm *ProcessedMessageIDsCache) Clean(timestamp uint64) error {
 	return pm.P.MessageCacheClearOlderThan(timestamp)
 }
+
+type WakuProtectedTopics struct {
+	P types.Persistence
+}
+
+var _ wakupersistence.ProtectedTopics = (*WakuProtectedTopics)(nil)
+
+func (wpt *WakuProtectedTopics) Insert(pubsubTopic string, privKey *ecdsa.PrivateKey, publicKey *ecdsa.PublicKey) error {
+	return wpt.P.WakuInsertProtectedTopic(pubsubTopic, privKey, publicKey)
+}
+
+func (wpt *WakuProtectedTopics) Delete(pubsubTopic string) error {
+	return wpt.P.WakuDeleteProtectedTopic(pubsubTopic)
+}
+
+func (wpt *WakuProtectedTopics) FetchPrivateKey(topic string) (*ecdsa.PrivateKey, error) {
+	return wpt.P.WakuFetchPrivateKeyForProtectedTopic(topic)
+}
+
+func (wpt *WakuProtectedTopics) ProtectedTopics() ([]wakupersistence.ProtectedTopic, error) {
+	pt, err := wpt.P.WakuProtectedTopics()
+	if err != nil {
+		return nil, err
+	}
+	result := make([]wakupersistence.ProtectedTopic, len(pt))
+	for i, p := range pt {
+		result[i] = wakupersistence.ProtectedTopic{
+			PubKey: p.PubKey,
+			Topic:  p.Topic,
+		}
+	}
+	return result, nil
+}

messaging/common/message_sender_test.go

@@ -77,7 +77,7 @@ func (s *MessageSenderSuite) SetupTest() {
 		nil,
 		&wakuConfig,
 		s.logger,
-		database,
+		nil,
 		nil,
 		func([]byte, peer.AddrInfo, error) {},
 		nil,

messaging/common/persistence_stub_test.go

@@ -29,6 +29,8 @@ type StubPersistence struct {
 	completedSegments map[string]struct{}                           // hash
 }
 
+var _ types.Persistence = (*StubPersistence)(nil)
+
 func NewStubPersistence() *StubPersistence {
 	return &StubPersistence{
 		wakuKeys:                   make(map[string][]byte),
@@ -244,21 +246,33 @@ func (s *StubPersistence) CompleteMessageSegments(hash []byte, sigPubKey *ecdsa.
 }
 
 func (s *StubPersistence) DeleteHashRatchetMessagesOlderThan(timestamp int64) error {
-	// Not implemented for stub
 	return nil
 }
 
 func (s *StubPersistence) InsertPendingConfirmation(*types.RawMessageConfirmation) error {
-	// Not implemented for stub
 	return nil
 }
 
 func (s *StubPersistence) RemoveMessageSegmentsOlderThan(timestamp int64) error {
-	// Not implemented for stub
 	return nil
 }
 
 func (s *StubPersistence) RemoveMessageSegmentsCompletedOlderThan(timestamp int64) error {
-	// Not implemented for stub
 	return nil
 }
+
+func (s *StubPersistence) WakuInsertProtectedTopic(pubsubTopic string, privKey *ecdsa.PrivateKey, publicKey *ecdsa.PublicKey) error {
+	return nil
+}
+
+func (s *StubPersistence) WakuDeleteProtectedTopic(pubsubTopic string) error {
+	return nil
+}
+
+func (s *StubPersistence) WakuFetchPrivateKeyForProtectedTopic(topic string) (*ecdsa.PrivateKey, error) {
+	return nil, nil
+}
+
+func (s *StubPersistence) WakuProtectedTopics() ([]types.ProtectedTopic, error) {
+	return nil, nil
+}

messaging/core.go

@@ -125,7 +125,7 @@ func NewCore(params CoreParams, options ...Options) (*Core, error) {
 	config := newConfig(options...)
 
 	waku, err := newWaku(wakuParams{
-		db:                              params.DB,
+		persistence:                     params.Persistence,
 		identity:                        params.Identity,
 		nodeKey:                         params.NodeKey,
 		wakuConfig:                      params.WakuConfig,
@@ -369,7 +369,7 @@ func (c *Core) startCleanupLoop(name string, cleanupFunc func() error) {
 }
 
 type wakuParams struct {
-	db *sql.DB
+	persistence types.Persistence
 
 	identity *ecdsa.PrivateKey
 	nodeKey  *ecdsa.PrivateKey
@@ -425,7 +425,7 @@ func newWaku(params wakuParams) (*wakuv2.Waku, error) {
 		params.nodeKey,
 		cfg,
 		params.logger,
-		params.db,
+		&adapters.WakuProtectedTopics{P: params.persistence},
 		params.timeSource,
 		params.onHistoricMessagesRequestFailed,
 		params.onPeerStats,

messaging/types/persistence.go

@@ -3,8 +3,7 @@ package types
 import "crypto/ecdsa"
 
 type Persistence interface {
-	WakuKeys() (map[string][]byte, error)
-	AddWakuKey(chatID string, key []byte) error
+	wakuPersistence
 
 	MessageCacheAdd(ids []string, timestamp uint64) error
 	MessageCacheClear() error
@@ -24,3 +23,17 @@ type Persistence interface {
 	RemoveMessageSegmentsOlderThan(timestamp int64) error
 	RemoveMessageSegmentsCompletedOlderThan(timestamp int64) error
 }
+
+type ProtectedTopic struct {
+	PubKey *ecdsa.PublicKey
+	Topic  string
+}
+
+type wakuPersistence interface {
+	WakuKeys() (map[string][]byte, error)
+	AddWakuKey(chatID string, key []byte) error
+	WakuInsertProtectedTopic(pubsubTopic string, privKey *ecdsa.PrivateKey, publicKey *ecdsa.PublicKey) error
+	WakuDeleteProtectedTopic(pubsubTopic string) error
+	WakuFetchPrivateKeyForProtectedTopic(topic string) (*ecdsa.PrivateKey, error)
+	WakuProtectedTopics() ([]ProtectedTopic, error)
+}

messaging/waku/gowaku.go

@@ -154,7 +154,7 @@ type Waku struct {
 
 	bandwidthCounter *metrics.BandwidthCounter
 
-	protectedTopicStore *persistence.ProtectedTopicsStore
+	protectedTopicStore persistence.ProtectedTopics
 
 	sendQueue *publish.MessageQueue
 
@@ -221,7 +221,7 @@ func newTTLCache() *ttlcache.Cache[gethcommon.Hash, bool] {
 }
 
 // New creates a WakuV2 client ready to communicate through the LibP2P network.
-func New(nodeKey *ecdsa.PrivateKey, cfg *Config, logger *zap.Logger, appDB *sql.DB, ts timesource.TimeSource, onHistoricMessagesRequestFailed func([]byte, peer.AddrInfo, error), onPeerStats func(types.ConnStatus)) (*Waku, error) {
+func New(nodeKey *ecdsa.PrivateKey, cfg *Config, logger *zap.Logger, protectedTopicsPersistence persistence.ProtectedTopics, ts timesource.TimeSource, onHistoricMessagesRequestFailed func([]byte, peer.AddrInfo, error), onPeerStats func(types.ConnStatus)) (*Waku, error) {
 	var err error
 	if logger == nil {
 		logger, err = zap.NewDevelopment()
@@ -244,7 +244,6 @@ func New(nodeKey *ecdsa.PrivateKey, cfg *Config, logger *zap.Logger, appDB *sql.
 	ctx, cancel := context.WithCancel(context.Background())
 
 	waku := &Waku{
-		appDB:                           appDB,
 		cfg:                             cfg,
 		privateKeys:                     make(map[string]*ecdsa.PrivateKey),
 		symKeys:                         make(map[string][]byte),
@@ -267,6 +266,7 @@ func New(nodeKey *ecdsa.PrivateKey, cfg *Config, logger *zap.Logger, appDB *sql.
 		onPeerStats:                     onPeerStats,
 		onlineChecker:                   onlinechecker.NewDefaultOnlineChecker(false).(*onlinechecker.DefaultOnlineChecker),
 		sendQueue:                       publish.NewMessageQueue(1000, cfg.UseThrottledPublish),
+		protectedTopicStore:             protectedTopicsPersistence,
 	}
 
 	waku.filters = common.NewFilters(waku.cfg.DefaultShardPubsubTopic, waku.logger)
@@ -348,13 +348,6 @@ func New(nodeKey *ecdsa.PrivateKey, cfg *Config, logger *zap.Logger, appDB *sql.
 		opts = append(opts, node.WithLightPush(lightpush.WithRateLimiter(5, 10)))
 	}
 
-	if appDB != nil {
-		waku.protectedTopicStore, err = persistence.NewProtectedTopicsStore(logger, appDB)
-		if err != nil {
-			return nil, err
-		}
-	}
-
 	if cfg.EnablePeerExchangeServer {
 		opts = append(opts, node.WithPeerExchange(peer_exchange.WithRateLimiter(1, 1)))
 	}
@@ -1399,13 +1392,6 @@ func (w *Waku) Stop() error {
 
 	w.node.Stop()
 
-	if w.protectedTopicStore != nil {
-		err := w.protectedTopicStore.Close()
-		if err != nil {
-			return err
-		}
-	}
-
 	close(w.goingOnline)
 	w.wg.Wait()
 

messaging/waku/nwaku.go

@@ -9,7 +9,6 @@ import (
 	"context"
 	"crypto/ecdsa"
 	"crypto/sha256"
-	"database/sql"
 	"encoding/hex"
 	"errors"
 	"fmt"
@@ -108,8 +107,6 @@ type IMetricsHandler interface {
 type Waku struct {
 	node *waku.WakuNode
 
-	appDB *sql.DB
-
 	dnsAddressCache             map[string][]dnsdisc.DiscoveredNode // Map to store the multiaddresses returned by dns discovery
 	dnsAddressCacheLock         *sync.RWMutex                       // lock to handle access to the map
 	dnsDiscAsyncRetrievedSignal chan struct{}
@@ -127,7 +124,7 @@ type Waku struct {
 
 	bandwidthCounter *metrics.BandwidthCounter
 
-	protectedTopicStore *persistence.ProtectedTopicsStore
+	protectedTopicStore persistence.ProtectedTopics
 
 	sendQueue *publish.MessageQueue
 
@@ -207,7 +204,7 @@ func newTTLCache() *ttlcache.Cache[gethcommon.Hash, bool] {
 }
 
 // New creates a WakuV2 client ready to communicate through the LibP2P network.
-func New(nodeKey *ecdsa.PrivateKey, cfg *Config, logger *zap.Logger, appDB *sql.DB, ts timesource.TimeSource, onHistoricMessagesRequestFailed func([]byte, peer.AddrInfo, error), onPeerStats func(types.ConnStatus)) (*Waku, error) {
+func New(nodeKey *ecdsa.PrivateKey, cfg *Config, logger *zap.Logger, protectedTopicsPersistence persistence.ProtectedTopics, ts timesource.TimeSource, onHistoricMessagesRequestFailed func([]byte, peer.AddrInfo, error), onPeerStats func(types.ConnStatus)) (*Waku, error) {
 	var err error
 	if logger == nil {
 		logger, err = zap.NewDevelopment()
@@ -257,7 +254,6 @@ func New(nodeKey *ecdsa.PrivateKey, cfg *Config, logger *zap.Logger, appDB *sql.
 
 	waku := &Waku{
 		node:                            wakunode,
-		appDB:                           appDB,
 		cfg:                             cfg,
 		privateKeys:                     make(map[string]*ecdsa.PrivateKey),
 		symKeys:                         make(map[string][]byte),
@@ -281,6 +277,7 @@ func New(nodeKey *ecdsa.PrivateKey, cfg *Config, logger *zap.Logger, appDB *sql.
 		onPeerStats:                     onPeerStats,
 		onlineChecker:                   onlinechecker.NewDefaultOnlineChecker(false).(*onlinechecker.DefaultOnlineChecker),
 		sendQueue:                       publish.NewMessageQueue(1000, cfg.UseThrottledPublish),
+		protectedTopicStore:             protectedTopicsPersistence,
 	}
 
 	waku.filters = common.NewFilters(waku.cfg.DefaultShardPubsubTopic, waku.logger)
@@ -1072,13 +1069,6 @@ func (w *Waku) Stop() error {
 		return err
 	}
 
-	if w.protectedTopicStore != nil {
-		err := w.protectedTopicStore.Close()
-		if err != nil {
-			return err
-		}
-	}
-
 	close(w.goingOnline)
 
 	w.wg.Wait()

messaging/waku/persistence/protected_topics.go

@@ -0,0 +1,17 @@
+package persistence
+
+import (
+	"crypto/ecdsa"
+)
+
+type ProtectedTopics interface {
+	Insert(pubsubTopic string, privKey *ecdsa.PrivateKey, publicKey *ecdsa.PublicKey) error
+	Delete(pubsubTopic string) error
+	FetchPrivateKey(topic string) (*ecdsa.PrivateKey, error)
+	ProtectedTopics() ([]ProtectedTopic, error)
+}
+
+type ProtectedTopic struct {
+	PubKey *ecdsa.PublicKey
+	Topic  string
+}