status-im
diff --git a/‎pkg/messaging/api_processor.go‎
Lines changed: 0 additions & 6 deletions b/‎pkg/messaging/api_processor.go‎
Lines changed: 0 additions & 6 deletions
diff --git a/‎pkg/messaging/layers/transport/filters_manager.go‎
Lines changed: 48 additions & 0 deletions b/‎pkg/messaging/layers/transport/filters_manager.go‎
Lines changed: 48 additions & 0 deletions
diff --git a/‎pkg/messaging/layers/transport/messaging_api.go‎
Lines changed: 6 additions & 3 deletions b/‎pkg/messaging/layers/transport/messaging_api.go‎
Lines changed: 6 additions & 3 deletions
@@ -3,8 +3,6 @@ package messaging
 import (
 	"crypto/ecdsa"
 
-	ethcommon "github.com/ethereum/go-ethereum/common"
-
 	"github.com/status-im/status-go/internal/connection"
 	adapters "github.com/status-im/status-go/pkg/messaging/adapters"
 	types "github.com/status-im/status-go/pkg/messaging/types"
@@ -38,10 +36,6 @@ func (a *API) ClearProcessedMessageIDsCache() error {
 	return a.core.stack.Transport.ClearProcessedMessageIDsCache()
 }
 
-func (a *API) MarkP2PMessageAsProcessed(hash ethcommon.Hash) {
-	a.core.stack.Transport.MarkP2PMessageAsProcessed(hash)
-}
-
 func (a *API) SetEnvelopeEventsHandler(handler types.EnvelopeEventsHandler) error {
 	return a.core.stack.Transport.SetEnvelopeEventsHandler(handler)
 }
 
@@ -47,6 +47,12 @@ type FiltersManager struct {
 	logger      *zap.Logger
 	mutex       sync.Mutex
 	filters     map[string]*Filter
+	// asymKeys maps an asymmetric filter's FilterID to the private key used to
+	// decode messages received on it. Symmetric keys are resolved on demand via
+	// keysManager.RawSymKey(SymKeyID) (the same path the send side uses), so only
+	// the asymmetric keys — which are not otherwise reachable from a persisted
+	// Filter — are held here. Runtime-only; never persisted.
+	asymKeys map[string]*ecdsa.PrivateKey
 }
 
 // NewFiltersManager returns a new filtersManager.
@@ -66,10 +72,47 @@ func NewFiltersManager(persistence KeysPersistence, service FiltersService, priv
 		persistence: persistence,
 		keys:        keys,
 		filters:     make(map[string]*Filter),
+		asymKeys:    make(map[string]*ecdsa.PrivateKey),
 		logger:      logger.With(zap.Namespace("filtersManager")),
 	}, nil
 }
 
+// WatchersByTopic returns the filters interested in the given (pubsubTopic,
+// contentTopic) pair. A filter with no explicit pubsub topic listens on the
+// default shard, so it is normalized before comparison — preserving the
+// (pubsub, content) routing the waku layer used. A single content topic can be
+// shared by distinct chats (content topics are a 4-byte hash of the chat id),
+// so the caller must still confirm the match by decoding with each candidate's
+// own key.
+func (f *FiltersManager) WatchersByTopic(pubsubTopic, contentTopic string) []*Filter {
+	f.mutex.Lock()
+	defer f.mutex.Unlock()
+
+	var res []*Filter
+	for _, filter := range f.filters {
+		if filter.ContentTopic.ContentTopic() != contentTopic {
+			continue
+		}
+		filterPubsubTopic := filter.PubsubTopic
+		if filterPubsubTopic == "" {
+			filterPubsubTopic = wakuv2.DefaultShardPubsubTopic()
+		}
+		if filterPubsubTopic == pubsubTopic {
+			res = append(res, filter)
+		}
+	}
+	return res
+}
+
+// AsymKey returns the asymmetric private key used to decode messages received
+// on the filter with the given FilterID, if it is an asymmetric filter.
+func (f *FiltersManager) AsymKey(filterID string) (*ecdsa.PrivateKey, bool) {
+	f.mutex.Lock()
+	defer f.mutex.Unlock()
+	k, ok := f.asymKeys[filterID]
+	return k, ok
+}
+
 func (f *FiltersManager) Init(
 	filtersToInit []FiltersToInitialize,
 	publicKeys []*ecdsa.PublicKey,
@@ -281,6 +324,7 @@ func (f *FiltersManager) Remove(ctx context.Context, filtersToRemove ...*Filter)
 		if filter.SymKeyID != "" {
 			f.service.DeleteSymKey(filter.SymKeyID)
 		}
+		delete(f.asymKeys, filter.FilterID)
 		for k, v := range f.filters {
 			if filter.FilterID == v.FilterID {
 				delete(f.filters, k)
@@ -312,6 +356,7 @@ func (f *FiltersManager) RemoveNoListenFilters() error {
 		if filter.SymKeyID != "" {
 			f.service.DeleteSymKey(filter.SymKeyID)
 		}
+		delete(f.asymKeys, filter.FilterID)
 		for k, v := range f.filters {
 			if filter.FilterID == v.FilterID {
 				delete(f.filters, k)
@@ -677,6 +722,9 @@ func (f *FiltersManager) addAsymmetric(chatID string, pubsubTopic string, identi
 	if err != nil {
 		return nil, err
 	}
+
+	f.asymKeys[id] = identity
+
 	return &RawFilter{FilterID: id, Topic: types.BytesToTopic(topic)}, nil
 }
 
 
@@ -18,9 +18,12 @@ type MessagingAPI interface {
 	// (see rfc26.Encode). Returns the wire hash on success.
 	Send(ctx context.Context, pubsubTopic, contentTopic string, payload []byte, ephemeral bool, priority *int) ([]byte, error)
 
-	// GetFilterMessages returns the messages that match the filter criteria
-	// and were received between the last poll and now.
-	GetFilterMessages(id string) ([]*types.Message, error)
+	// SubscribeEnvelopeEvents returns the backend's event stream. The transport
+	// consumes it both for reception (EventEnvelopeAvailable carries a neutral
+	// *types.ReceivedMessage in Data, which the transport decodes and routes to
+	// its filters) and for the send-side lifecycle events the EnvelopesMonitor
+	// tracks. This is the single, logos-delivery-shaped event seam.
+	SubscribeEnvelopeEvents(events chan<- types.EnvelopeEvent) types.Subscription
 }
 
 // Compile-time assertion: the waku adapter satisfies MessagingAPI.