Repository: hyperledger/firefly
Pull request #911 (2022-07-27 16:10:11)

Properly handle remote->local namespace mapping in blockchain plugin
In the "new world" (network version 2), the blockchain plugin (like other plugins) should only deal with the local namespace name.

For network version 1, the (remote) namespace must be written on chain, so it's necessary to map back and forth from local to remote namespace name within the plugin. This does break the paradigm of keeping namespace mapping at a higher level only, but seems like the cleanest solution for old environments.

Note that this also alters the subscription creation logic for V1 vs. V2. For V1, there will be a shared subscription for all namespaces on a given contract (which is how FireFly 1.0 already behaves - it was changed in #865, but it's actually easiest to keep the old behavior for V1 contracts). For V2+ only, there will be a separate subscription created per namespace.

Includes a basic E2E test for namespace mapping.


Patch:
diff --git a/internal/blockchain/common/common.go b/internal/blockchain/common/common.go
index 836c75a43..2de04f288 100644
--- a/internal/blockchain/common/common.go
+++ b/internal/blockchain/common/common.go
@@ -33,17 +33,43 @@ type BlockchainCallbacks interface {
 	SetHandler(namespace string, handler blockchain.Callbacks)
 	SetOperationalHandler(namespace string, handler core.OperationCallbacks)
 
-	OperationUpdate(ctx context.Context, plugin blockchain.Plugin, nsOpID string, status core.OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject)
-	BatchPinComplete(ctx context.Context, batch *blockchain.BatchPin, signingKey *core.VerifierRef) error
-	BlockchainNetworkAction(ctx context.Context, namespace, action string, location *fftypes.JSONAny, event *blockchain.Event, signingKey *core.VerifierRef) error
+	OperationUpdate(ctx context.Context, plugin core.Named, nsOpID string, status core.OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject)
+	BatchPinOrNetworkAction(ctx context.Context, nsOrAction string, subInfo *SubscriptionInfo, location *fftypes.JSONAny, event *blockchain.Event, signingKey *core.VerifierRef, params *BatchPinParams) error
 	BlockchainEvent(ctx context.Context, namespace string, event *blockchain.EventWithSubscription) error
 }
 
+type FireflySubscriptions interface {
+	AddSubscription(ctx context.Context, namespace core.NamespaceRef, version int, subID string, extra interface{})
+	RemoveSubscription(ctx context.Context, subID string)
+	GetSubscription(subID string) *SubscriptionInfo
+}
+
 type callbacks struct {
 	handlers   map[string]blockchain.Callbacks
 	opHandlers map[string]core.OperationCallbacks
 }
 
+type subscriptions struct {
+	subs map[string]*SubscriptionInfo
+}
+
+type BatchPinParams struct {
+	UUIDs      string
+	BatchHash  string
+	Contexts   []string
+	PayloadRef string
+}
+
+// A single subscription on network version 1 may receive events from many remote namespaces,
+// which in turn map to one or more local namespaces.
+// A subscription on network version 2 is always specific to a single local namespace.
+type SubscriptionInfo struct {
+	Version     int
+	V1Namespace map[string][]string
+	V2Namespace string
+	Extra       interface{}
+}
+
 func NewBlockchainCallbacks() BlockchainCallbacks {
 	return &callbacks{
 		handlers:   make(map[string]blockchain.Callbacks),
@@ -51,6 +77,12 @@ func NewBlockchainCallbacks() BlockchainCallbacks {
 	}
 }
 
+func NewFireflySubscriptions() FireflySubscriptions {
+	return &subscriptions{
+		subs: make(map[string]*SubscriptionInfo),
+	}
+}
+
 func (cb *callbacks) SetHandler(namespace string, handler blockchain.Callbacks) {
 	cb.handlers[namespace] = handler
 }
@@ -59,7 +91,7 @@ func (cb *callbacks) SetOperationalHandler(namespace string, handler core.Operat
 	cb.opHandlers[namespace] = handler
 }
 
-func (cb *callbacks) OperationUpdate(ctx context.Context, plugin blockchain.Plugin, nsOpID string, status core.OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject) {
+func (cb *callbacks) OperationUpdate(ctx context.Context, plugin core.Named, nsOpID string, status core.OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject) {
 	namespace, _, _ := core.ParseNamespacedOpID(ctx, nsOpID)
 	if handler, ok := cb.opHandlers[namespace]; ok {
 		handler.OperationUpdate(plugin, nsOpID, status, blockchainTXID, errorMessage, opOutput)
@@ -68,36 +100,71 @@ func (cb *callbacks) OperationUpdate(ctx context.Context, plugin blockchain.Plug
 	log.L(ctx).Errorf("No handler found for blockchain operation '%s'", nsOpID)
 }
 
-func (cb *callbacks) BatchPinComplete(ctx context.Context, batch *blockchain.BatchPin, signingKey *core.VerifierRef) error {
-	if handler, ok := cb.handlers[batch.Namespace]; ok {
-		return handler.BatchPinComplete(batch, signingKey)
+func (cb *callbacks) BatchPinOrNetworkAction(ctx context.Context, nsOrAction string, subInfo *SubscriptionInfo, location *fftypes.JSONAny, event *blockchain.Event, signingKey *core.VerifierRef, params *BatchPinParams) error {
+	// Check if this is actually an operator action
+	if strings.HasPrefix(nsOrAction, blockchain.FireFlyActionPrefix) {
+		action := nsOrAction[len(blockchain.FireFlyActionPrefix):]
+
+		// For V1 of the FireFly contract, action is sent to all namespaces.
+		// For V2+, namespace is inferred from the subscription.
+		if subInfo.Version == 1 {
+			namespaces := make([]string, 0)
+			for _, localNames := range subInfo.V1Namespace {
+				namespaces = append(namespaces, localNames...)
+			}
+			return cb.networkAction(ctx, namespaces, action, location, event, signingKey)
+		}
+		return cb.networkAction(ctx, []string{subInfo.V2Namespace}, action, location, event, signingKey)
 	}
-	log.L(ctx).Errorf("No handler found for blockchain batch pin on namespace '%s'", batch.Namespace)
-	return nil
+
+	batch, err := buildBatchPin(ctx, event, params)
+	if err != nil {
+		return nil // move on
+	}
+
+	// For V1 of the FireFly contract, namespace is passed explicitly, but needs to be mapped to local name(s).
+	// For V2+, namespace is inferred from the subscription.
+	if subInfo.Version == 1 {
+		namespaces := subInfo.V1Namespace[nsOrAction]
+		if len(namespaces) == 0 {
+			log.L(ctx).Errorf("No handler found for blockchain batch pin on remote namespace '%s'", nsOrAction)
+			return nil
+		}
+		return cb.batchPinComplete(ctx, namespaces, batch, signingKey)
+	}
+	return cb.batchPinComplete(ctx, []string{subInfo.V2Namespace}, batch, signingKey)
 }
 
-func (cb *callbacks) BlockchainNetworkAction(ctx context.Context, namespace, action string, location *fftypes.JSONAny, event *blockchain.Event, signingKey *core.VerifierRef) error {
-	if namespace == "" {
-		// V1 networks don't populate namespace, so deliver the event to every handler
-		for _, handler := range cb.handlers {
-			if err := handler.BlockchainNetworkAction(action, location, event, signingKey); err != nil {
+func (cb *callbacks) batchPinComplete(ctx context.Context, namespaces []string, batch *blockchain.BatchPin, signingKey *core.VerifierRef) error {
+	for _, namespace := range namespaces {
+		if handler, ok := cb.handlers[namespace]; ok {
+			if err := handler.BatchPinComplete(namespace, batch, signingKey); err != nil {
 				return err
 			}
+		} else {
+			log.L(ctx).Errorf("No handler found for blockchain batch pin on local namespace '%s'", namespace)
 		}
-	} else {
+	}
+	return nil
+}
+
+func (cb *callbacks) networkAction(ctx context.Context, namespaces []string, action string, location *fftypes.JSONAny, event *blockchain.Event, signingKey *core.VerifierRef) error {
+	for _, namespace := range namespaces {
 		if handler, ok := cb.handlers[namespace]; ok {
-			return handler.BlockchainNetworkAction(action, location, event, signingKey)
+			if err := handler.BlockchainNetworkAction(action, location, event, signingKey); err != nil {
+				return err
+			}
+		} else {
+			log.L(ctx).Errorf("No handler found for blockchain network action on local namespace '%s'", namespace)
 		}
-		log.L(ctx).Errorf("No handler found for blockchain network action on namespace '%s'", namespace)
 	}
 	return nil
 }
 
 func (cb *callbacks) BlockchainEvent(ctx context.Context, namespace string, event *blockchain.EventWithSubscription) error {
 	if namespace == "" {
-		// Older token subscriptions don't populate namespace, so deliver the event to every handler
+		// Older subscriptions don't populate namespace, so deliver the event to every handler
 		for _, cb := range cb.handlers {
-			// Send the event to all handlers and let them match it to a contract listener
 			if err := cb.BlockchainEvent(event); err != nil {
 				return err
 			}
@@ -111,11 +178,16 @@ func (cb *callbacks) BlockchainEvent(ctx context.Context, namespace string, even
 	return nil
 }
 
-func BuildBatchPin(ctx context.Context, namespace string, event *blockchain.Event, sUUIDs string, sBatchHash string, sContexts []string, sPayloadRef string) (batch *blockchain.BatchPin, err error) {
-	hexUUIDs, err := hex.DecodeString(strings.TrimPrefix(sUUIDs, "0x"))
+func buildBatchPin(ctx context.Context, event *blockchain.Event, params *BatchPinParams) (batch *blockchain.BatchPin, err error) {
+	if params.UUIDs == "" || params.BatchHash == "" {
+		log.L(ctx).Errorf("BatchPin event is not valid - missing data: %+v", params)
+		return nil, i18n.NewError(ctx, coremsgs.MsgInvalidBatchPinEvent, "missing data", params.UUIDs, "")
+	}
+
+	hexUUIDs, err := hex.DecodeString(strings.TrimPrefix(params.UUIDs, "0x"))
 	if err != nil || len(hexUUIDs) != 32 {
-		log.L(ctx).Errorf("BatchPin event is not valid - bad uuids (%s): %s", sUUIDs, err)
-		return nil, i18n.NewError(ctx, coremsgs.MsgInvalidBatchPinEvent, "bad uuids", sUUIDs, err)
+		log.L(ctx).Errorf("BatchPin event is not valid - bad uuids (%s): %s", params.UUIDs, err)
+		return nil, i18n.NewError(ctx, coremsgs.MsgInvalidBatchPinEvent, "bad uuids", params.UUIDs, err)
 	}
 	var txnID fftypes.UUID
 	copy(txnID[:], hexUUIDs[0:16])
@@ -123,14 +195,14 @@ func BuildBatchPin(ctx context.Context, namespace string, event *blockchain.Even
 	copy(batchID[:], hexUUIDs[16:32])
 
 	var batchHash fftypes.Bytes32
-	err = batchHash.UnmarshalText([]byte(sBatchHash))
+	err = batchHash.UnmarshalText([]byte(params.BatchHash))
 	if err != nil {
-		log.L(ctx).Errorf("BatchPin event is not valid - bad batchHash (%s): %s", sBatchHash, err)
+		log.L(ctx).Errorf("BatchPin event is not valid - bad batchHash (%s): %s", params.BatchHash, err)
 		return nil, err
 	}
 
-	contexts := make([]*fftypes.Bytes32, len(sContexts))
-	for i, sHash := range sContexts {
+	contexts := make([]*fftypes.Bytes32, len(params.Contexts))
+	for i, sHash := range params.Contexts {
 		var hash fftypes.Bytes32
 		err = hash.UnmarshalText([]byte(sHash))
 		if err != nil {
@@ -141,12 +213,58 @@ func BuildBatchPin(ctx context.Context, namespace string, event *blockchain.Even
 	}
 
 	return &blockchain.BatchPin{
-		Namespace:       namespace,
 		TransactionID:   &txnID,
 		BatchID:         &batchID,
 		BatchHash:       &batchHash,
-		BatchPayloadRef: sPayloadRef,
+		BatchPayloadRef: params.PayloadRef,
 		Contexts:        contexts,
 		Event:           *event,
 	}, nil
 }
+
+func GetNamespaceFromSubName(subName string) string {
+	var parts = strings.Split(subName, "-")
+	// Subscription names post version 1.1 are in the format `ff-sub-<namespace>-<listener ID>`
+	if len(parts) != 4 {
+		// Assume older subscription and return empty string
+		return ""
+	}
+	return parts[2]
+}
+
+func (s *subscriptions) AddSubscription(ctx context.Context, namespace core.NamespaceRef, version int, subID string, extra interface{}) {
+	if version == 1 {
+		// The V1 contract shares a single subscription per contract, and the remote namespace name is passed on chain.
+		// Therefore, it requires a map of remote->local in order to farm out notifications to one or more local handlers.
+		existing, ok := s.subs[subID]
+		if !ok {
+			existing = &SubscriptionInfo{
+				Version:     version,
+				V1Namespace: make(map[string][]string),
+				Extra:       extra,
+			}
+			s.subs[subID] = existing
+		}
+		existing.V1Namespace[namespace.RemoteName] = append(existing.V1Namespace[namespace.RemoteName], namespace.LocalName)
+	} else {
+		// The V2 contract does not pass the namespace on chain, and requires a separate contract instance (and subscription) per namespace.
+		// Therefore, the local namespace name can simply be cached alongside each subscription.
+		s.subs[subID] = &SubscriptionInfo{
+			Version:     version,
+			V2Namespace: namespace.LocalName,
+			Extra:       extra,
+		}
+	}
+}
+
+func (s *subscriptions) RemoveSubscription(ctx context.Context, subID string) {
+	if _, ok := s.subs[subID]; ok {
+		delete(s.subs, subID)
+	} else {
+		log.L(ctx).Debugf("Invalid subscription ID: %s", subID)
+	}
+}
+
+func (s *subscriptions) GetSubscription(subID string) *SubscriptionInfo {
+	return s.subs[subID]
+}
diff --git a/internal/blockchain/common/eventstream.go b/internal/blockchain/common/eventstream.go
deleted file mode 100644
index ec0f09a6c..000000000
--- a/internal/blockchain/common/eventstream.go
+++ /dev/null
@@ -1,29 +0,0 @@
-// Copyright © 2022 Kaleido, Inc.
-//
-// SPDX-License-Identifier: Apache-2.0
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package common
-
-import "strings"
-
-func GetNamespaceFromSubName(subName string) string {
-	var parts = strings.Split(subName, "-")
-	// Subscription names post version 1.1 are in the format `ff-sub-<namespace>-<listener ID>`
-	if len(parts) != 4 {
-		// Assume older subscription and return empty string
-		return ""
-	}
-	return parts[2]
-}
diff --git a/internal/blockchain/ethereum/ethereum.go b/internal/blockchain/ethereum/ethereum.go
index 32a980e0c..d7710304a 100644
--- a/internal/blockchain/ethereum/ethereum.go
+++ b/internal/blockchain/ethereum/ethereum.go
@@ -64,16 +64,11 @@ type Ethereum struct {
 	addressResolver *addressResolver
 	metrics         metrics.Manager
 	ethconnectConf  config.Section
-	subs            map[string]subscriptionInfo
+	subs            common.FireflySubscriptions
 	cache           *ccache.Cache
 	cacheTTL        time.Duration
 }
 
-type subscriptionInfo struct {
-	namespace string
-	version   int
-}
-
 type eventStreamWebsocket struct {
 	Topic string `json:"topic"`
 }
@@ -132,6 +127,7 @@ func (e *Ethereum) Init(ctx context.Context, conf config.Section, metrics metric
 	e.metrics = metrics
 	e.capabilities = &blockchain.Capabilities{}
 	e.callbacks = common.NewBlockchainCallbacks()
+	e.subs = common.NewFireflySubscriptions()
 
 	if addressResolverConf.GetString(AddressResolverURLTemplate) != "" {
 		if e.addressResolver, err = newAddressResolver(ctx, addressResolverConf); err != nil {
@@ -175,7 +171,6 @@ func (e *Ethereum) Init(ctx context.Context, conf config.Section, metrics metric
 		return err
 	}
 	e.streamID = stream.ID
-	e.subs = make(map[string]subscriptionInfo)
 	log.L(e.ctx).Infof("Event stream: %s (topic=%s)", e.streamID, e.topic)
 
 	e.closed = make(chan struct{})
@@ -200,7 +195,7 @@ func (e *Ethereum) Capabilities() *blockchain.Capabilities {
 	return e.capabilities
 }
 
-func (e *Ethereum) AddFireflySubscription(ctx context.Context, namespace string, location *fftypes.JSONAny, firstEvent string) (string, error) {
+func (e *Ethereum) AddFireflySubscription(ctx context.Context, namespace core.NamespaceRef, location *fftypes.JSONAny, firstEvent string) (string, error) {
 	ethLocation, err := parseContractLocation(ctx, location)
 	if err != nil {
 		return "", err
@@ -213,24 +208,17 @@ func (e *Ethereum) AddFireflySubscription(ctx context.Context, namespace string,
 		firstEvent = "latest"
 	}
 
-	sub, subNS, err := e.streams.ensureFireFlySubscription(ctx, namespace, ethLocation.Address, firstEvent, e.streamID, batchPinEventABI)
+	version, err := e.GetNetworkVersion(ctx, location)
 	if err != nil {
 		return "", err
 	}
 
-	version, err := e.GetNetworkVersion(ctx, location)
+	sub, err := e.streams.ensureFireFlySubscription(ctx, namespace.LocalName, version, ethLocation.Address, firstEvent, e.streamID, batchPinEventABI)
 	if err != nil {
 		return "", err
 	}
 
-	if version > 1 && subNS == "" {
-		return "", i18n.NewError(ctx, coremsgs.MsgInvalidSubscriptionForNetwork, sub.Name, version)
-	}
-
-	e.subs[sub.ID] = subscriptionInfo{
-		namespace: subNS,
-		version:   version,
-	}
+	e.subs.AddSubscription(ctx, namespace, version, sub.ID, nil)
 	return sub.ID, nil
 }
 
@@ -238,11 +226,7 @@ func (e *Ethereum) RemoveFireflySubscription(ctx context.Context, subID string)
 	// Don't actually delete the subscription from ethconnect, as this may be called while processing
 	// events from the subscription (and handling that scenario cleanly could be difficult for ethconnect).
 	// TODO: can old subscriptions be somehow cleaned up later?
-	if _, ok := e.subs[subID]; ok {
-		delete(e.subs, subID)
-	} else {
-		log.L(ctx).Debugf("Invalid subscription ID: %s", subID)
-	}
+	e.subs.RemoveSubscription(ctx, subID)
 }
 
 func (e *Ethereum) afterConnect(ctx context.Context, w wsclient.WSClient) error {
@@ -303,7 +287,7 @@ func (e *Ethereum) parseBlockchainEvent(ctx context.Context, msgJSON fftypes.JSO
 	}
 }
 
-func (e *Ethereum) handleBatchPinEvent(ctx context.Context, location *fftypes.JSONAny, subInfo *subscriptionInfo, msgJSON fftypes.JSONObject) (err error) {
+func (e *Ethereum) handleBatchPinEvent(ctx context.Context, location *fftypes.JSONAny, subInfo *common.SubscriptionInfo, msgJSON fftypes.JSONObject) (err error) {
 	event := e.parseBlockchainEvent(ctx, msgJSON)
 	if event == nil {
 		return nil // move on
@@ -314,14 +298,12 @@ func (e *Ethereum) handleBatchPinEvent(ctx context.Context, location *fftypes.JS
 	if nsOrAction == "" {
 		nsOrAction = event.Output.GetString("namespace")
 	}
-	sUUIDs := event.Output.GetString("uuids")
-	sBatchHash := event.Output.GetString("batchHash")
-	sPayloadRef := event.Output.GetString("payloadRef")
-	sContexts := event.Output.GetStringArray("contexts")
 
-	if authorAddress == "" || sUUIDs == "" || sBatchHash == "" {
-		log.L(ctx).Errorf("BatchPin event is not valid - missing data: %+v", msgJSON)
-		return nil // move on
+	params := &common.BatchPinParams{
+		UUIDs:      event.Output.GetString("uuids"),
+		BatchHash:  event.Output.GetString("batchHash"),
+		PayloadRef: event.Output.GetString("payloadRef"),
+		Contexts:   event.Output.GetStringArray("contexts"),
 	}
 
 	authorAddress, err = e.NormalizeSigningKey(ctx, authorAddress)
@@ -334,40 +316,7 @@ func (e *Ethereum) handleBatchPinEvent(ctx context.Context, location *fftypes.JS
 		Value: authorAddress,
 	}
 
-	// Check if this is actually an operator action
-	if strings.HasPrefix(nsOrAction, blockchain.FireFlyActionPrefix) {
-		action := nsOrAction[len(blockchain.FireFlyActionPrefix):]
-
-		// For V1 of the FireFly contract, action is sent to all namespaces
-		// For V2+, namespace is inferred from the subscription
-		var namespace string
-		if subInfo.version > 1 {
-			namespace = subInfo.namespace
-		}
-
-		return e.callbacks.BlockchainNetworkAction(ctx, namespace, action, location, event, verifier)
-	}
-
-	// For V1 of the FireFly contract, namespace is passed explicitly
-	// For V2+, namespace is inferred from the subscription
-	var namespace string
-	if subInfo.version == 1 {
-		namespace = nsOrAction
-		if subInfo.namespace != "" && subInfo.namespace != namespace {
-			log.L(ctx).Debugf("Ignoring batch for '%s' received on subscription for '%s'", namespace, subInfo.namespace)
-			return nil
-		}
-	} else {
-		namespace = subInfo.namespace
-	}
-
-	batch, err := common.BuildBatchPin(ctx, namespace, event, sUUIDs, sBatchHash, sContexts, sPayloadRef)
-	if err != nil {
-		return nil // move on
-	}
-
-	// If there's an error dispatching the event, we must return the error and shutdown
-	return e.callbacks.BatchPinComplete(ctx, batch, verifier)
+	return e.callbacks.BatchPinOrNetworkAction(ctx, nsOrAction, subInfo, location, event, verifier, params)
 }
 
 func (e *Ethereum) handleContractEvent(ctx context.Context, msgJSON fftypes.JSONObject) (err error) {
@@ -429,7 +378,7 @@ func (e *Ethereum) handleMessageBatch(ctx context.Context, messages []interface{
 		logger.Tracef("Message: %+v", msgJSON)
 
 		// Matches one of the active FireFly BatchPin subscriptions
-		if subInfo, ok := e.subs[sub]; ok {
+		if subInfo := e.subs.GetSubscription(sub); subInfo != nil {
 			location, err := encodeContractLocation(ctx, &Location{
 				Address: msgJSON.GetString("address"),
 			})
@@ -440,7 +389,7 @@ func (e *Ethereum) handleMessageBatch(ctx context.Context, messages []interface{
 
 			switch signature {
 			case broadcastBatchEventSignature:
-				if err := e.handleBatchPinEvent(eventCtx, location, &subInfo, msgJSON); err != nil {
+				if err := e.handleBatchPinEvent(eventCtx, location, subInfo, msgJSON); err != nil {
 					done()
 					return err
 				}
@@ -607,7 +556,7 @@ func (e *Ethereum) queryContractMethod(ctx context.Context, address string, abi
 	return res, nil
 }
 
-func (e *Ethereum) SubmitBatchPin(ctx context.Context, nsOpID string, signingKey string, batch *blockchain.BatchPin, location *fftypes.JSONAny) error {
+func (e *Ethereum) SubmitBatchPin(ctx context.Context, nsOpID, remoteNamespace, signingKey string, batch *blockchain.BatchPin, location *fftypes.JSONAny) error {
 	ethLocation, err := parseContractLocation(ctx, location)
 	if err != nil {
 		return err
@@ -632,7 +581,7 @@ func (e *Ethereum) SubmitBatchPin(ctx context.Context, nsOpID string, signingKey
 	if version == 1 {
 		method = batchPinMethodABIV1
 		input = []interface{}{
-			batch.Namespace,
+			remoteNamespace,
 			ethHexFormatB32(&uuids),
 			ethHexFormatB32(batch.BatchHash),
 			batch.BatchPayloadRef,
diff --git a/internal/blockchain/ethereum/eventstream.go b/internal/blockchain/ethereum/eventstream.go
index e4aaa9db2..a5ad9ed0b 100644
--- a/internal/blockchain/ethereum/eventstream.go
+++ b/internal/blockchain/ethereum/eventstream.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/go-resty/resty/v2"
 	"github.com/hyperledger/firefly-common/pkg/ffresty"
+	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly-signer/pkg/abi"
 	"github.com/hyperledger/firefly/internal/coremsgs"
@@ -209,7 +210,7 @@ func (s *streamManager) deleteSubscription(ctx context.Context, subID string) er
 	return nil
 }
 
-func (s *streamManager) ensureFireFlySubscription(ctx context.Context, namespace string, instancePath, fromBlock, stream string, abi *abi.Entry) (sub *subscription, subNS string, err error) {
+func (s *streamManager) ensureFireFlySubscription(ctx context.Context, namespace string, version int, instancePath, fromBlock, stream string, abi *abi.Entry) (sub *subscription, err error) {
 	// Include a hash of the instance path in the subscription, so if we ever point at a different
 	// contract configuration, we re-subscribe from block 0.
 	// We don't need full strength hashing, so just use the first 16 chars for readability.
@@ -217,39 +218,43 @@ func (s *streamManager) ensureFireFlySubscription(ctx context.Context, namespace
 
 	existingSubs, err := s.getSubscriptions(ctx)
 	if err != nil {
-		return nil, "", err
+		return nil, err
 	}
 
-	oldName1 := abi.Name
-	oldName2 := fmt.Sprintf("%s_%s", abi.Name, instanceUniqueHash)
-	currentName := fmt.Sprintf("%s_%s_%s", namespace, abi.Name, instanceUniqueHash)
+	legacyName := abi.Name
+	v1Name := fmt.Sprintf("%s_%s", abi.Name, instanceUniqueHash)
+	v2Name := fmt.Sprintf("%s_%s_%s", namespace, abi.Name, instanceUniqueHash)
 
 	for _, s := range existingSubs {
 		if s.Stream == stream {
 			/* Check for the deprecated names, before adding namespace uniqueness qualifier.
 			   NOTE: If one of these early environments needed a new subscription, the existing one would need to
 				 be deleted manually. */
-			if s.Name == oldName1 || s.Name == oldName2 {
-				log.L(ctx).Warnf("Subscription %s uses a legacy name format '%s' and may not support multiple namespaces. Expected '%s' instead.", s.ID, s.Name, currentName)
-				sub = s
-			} else if s.Name == currentName {
-				sub = s
-				subNS = namespace
+			if version == 1 {
+				if s.Name == legacyName {
+					log.L(ctx).Warnf("Subscription %s uses a legacy name format '%s' - expected '%s' instead", s.ID, legacyName, v1Name)
+					return s, nil
+				} else if s.Name == v1Name {
+					return s, nil
+				}
+			} else {
+				if s.Name == legacyName || s.Name == v1Name {
+					return nil, i18n.NewError(ctx, coremsgs.MsgInvalidSubscriptionForNetwork, s.Name, version)
+				} else if s.Name == v2Name {
+					return s, nil
+				}
 			}
 		}
 	}
 
-	location := &Location{
-		Address: instancePath,
+	name := v2Name
+	if version == 1 {
+		name = v1Name
 	}
-
-	if sub == nil {
-		if sub, err = s.createSubscription(ctx, location, stream, currentName, fromBlock, abi); err != nil {
-			return nil, "", err
-		}
-		subNS = namespace
+	location := &Location{Address: instancePath}
+	if sub, err = s.createSubscription(ctx, location, stream, name, fromBlock, abi); err != nil {
+		return nil, err
 	}
-
 	log.L(ctx).Infof("%s subscription: %s", abi.Name, sub.ID)
-	return sub, subNS, nil
+	return sub, nil
 }
diff --git a/internal/blockchain/fabric/eventstream.go b/internal/blockchain/fabric/eventstream.go
index e538524c6..f26028fa8 100644
--- a/internal/blockchain/fabric/eventstream.go
+++ b/internal/blockchain/fabric/eventstream.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/go-resty/resty/v2"
 	"github.com/hyperledger/firefly-common/pkg/ffresty"
+	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/pkg/core"
@@ -190,34 +191,38 @@ func (s *streamManager) deleteSubscription(ctx context.Context, subID string) er
 	return nil
 }
 
-func (s *streamManager) ensureFireFlySubscription(ctx context.Context, namespace string, location *Location, fromBlock, stream, event string) (sub *subscription, subNS string, err error) {
+func (s *streamManager) ensureFireFlySubscription(ctx context.Context, namespace string, version int, location *Location, fromBlock, stream, event string) (sub *subscription, err error) {
 	existingSubs, err := s.getSubscriptions(ctx)
 	if err != nil {
-		return nil, "", err
+		return nil, err
 	}
 
-	oldName := event
-	currentName := fmt.Sprintf("%s_%s", namespace, event)
+	v1Name := event
+	v2Name := fmt.Sprintf("%s_%s", namespace, event)
 
 	for _, s := range existingSubs {
 		if s.Stream == stream {
-			if s.Name == oldName {
-				log.L(ctx).Warnf("Subscription %s uses a legacy name format '%s' and may not support multiple namespaces. Expected '%s' instead.", s.ID, s.Name, currentName)
-				sub = s
-			} else if s.Name == currentName {
-				sub = s
-				subNS = namespace
+			if version == 1 {
+				if s.Name == v1Name {
+					return s, nil
+				}
+			} else {
+				if s.Name == v1Name {
+					return nil, i18n.NewError(ctx, coremsgs.MsgInvalidSubscriptionForNetwork, s.Name, version)
+				} else if s.Name == v2Name {
+					return s, nil
+				}
 			}
 		}
 	}
 
-	if sub == nil {
-		if sub, err = s.createSubscription(ctx, location, stream, currentName, event, fromBlock); err != nil {
-			return nil, "", err
-		}
-		subNS = namespace
+	name := v2Name
+	if version == 1 {
+		name = v1Name
+	}
+	if sub, err = s.createSubscription(ctx, location, stream, name, event, fromBlock); err != nil {
+		return nil, err
 	}
-
 	log.L(ctx).Infof("%s subscription: %s", event, sub.ID)
-	return sub, subNS, nil
+	return sub, nil
 }
diff --git a/internal/blockchain/fabric/fabric.go b/internal/blockchain/fabric/fabric.go
index 263b87e8f..0e6c2b364 100644
--- a/internal/blockchain/fabric/fabric.go
+++ b/internal/blockchain/fabric/fabric.go
@@ -63,17 +63,11 @@ type Fabric struct {
 	closed         chan struct{}
 	metrics        metrics.Manager
 	fabconnectConf config.Section
-	subs           map[string]subscriptionInfo
+	subs           common.FireflySubscriptions
 	cache          *ccache.Cache
 	cacheTTL       time.Duration
 }
 
-type subscriptionInfo struct {
-	namespace string
-	channel   string
-	version   int
-}
-
 type eventStreamWebsocket struct {
 	Topic string `json:"topic"`
 }
@@ -202,6 +196,7 @@ func (f *Fabric) Init(ctx context.Context, conf config.Section, metrics metrics.
 	f.metrics = metrics
 	f.capabilities = &blockchain.Capabilities{}
 	f.callbacks = common.NewBlockchainCallbacks()
+	f.subs = common.NewFireflySubscriptions()
 
 	if fabconnectConf.GetString(ffresty.HTTPConfigURL) == "" {
 		return i18n.NewError(ctx, coremsgs.MsgMissingPluginConfig, "url", "blockchain.fabric.fabconnect")
@@ -238,7 +233,6 @@ func (f *Fabric) Init(ctx context.Context, conf config.Section, metrics metrics.
 		return err
 	}
 	f.streamID = stream.ID
-	f.subs = make(map[string]subscriptionInfo)
 	log.L(f.ctx).Infof("Event stream: %s", f.streamID)
 
 	f.closed = make(chan struct{})
@@ -323,7 +317,7 @@ func (f *Fabric) parseBlockchainEvent(ctx context.Context, msgJSON fftypes.JSONO
 	}
 }
 
-func (f *Fabric) handleBatchPinEvent(ctx context.Context, location *fftypes.JSONAny, subInfo *subscriptionInfo, msgJSON fftypes.JSONObject) (err error) {
+func (f *Fabric) handleBatchPinEvent(ctx context.Context, location *fftypes.JSONAny, subInfo *common.SubscriptionInfo, msgJSON fftypes.JSONObject) (err error) {
 	event := f.parseBlockchainEvent(ctx, msgJSON)
 	if event == nil {
 		return nil // move on
@@ -331,50 +325,19 @@ func (f *Fabric) handleBatchPinEvent(ctx context.Context, location *fftypes.JSON
 
 	signer := event.Output.GetString("signer")
 	nsOrAction := event.Output.GetString("namespace")
-	sUUIDs := event.Output.GetString("uuids")
-	sBatchHash := event.Output.GetString("batchHash")
-	sPayloadRef := event.Output.GetString("payloadRef")
-	sContexts := event.Output.GetStringArray("contexts")
+	params := &common.BatchPinParams{
+		UUIDs:      event.Output.GetString("uuids"),
+		BatchHash:  event.Output.GetString("batchHash"),
+		PayloadRef: event.Output.GetString("payloadRef"),
+		Contexts:   event.Output.GetStringArray("contexts"),
+	}
 
 	verifier := &core.VerifierRef{
 		Type:  core.VerifierTypeMSPIdentity,
 		Value: signer,
 	}
 
-	// Check if this is actually an operator action
-	if strings.HasPrefix(nsOrAction, blockchain.FireFlyActionPrefix) {
-		action := nsOrAction[len(blockchain.FireFlyActionPrefix):]
-
-		// For V1 of the FireFly contract, action is sent to all namespaces
-		// For V2+, namespace is inferred from the subscription
-		var namespace string
-		if subInfo.version > 1 {
-			namespace = subInfo.namespace
-		}
-
-		return f.callbacks.BlockchainNetworkAction(ctx, namespace, action, location, event, verifier)
-	}
-
-	// For V1 of the FireFly contract, namespace is passed explicitly
-	// For V2+, namespace is inferred from the subscription
-	var namespace string
-	if subInfo.version == 1 {
-		namespace = nsOrAction
-		if subInfo.namespace != "" && subInfo.namespace != namespace {
-			log.L(ctx).Debugf("Ignoring batch for '%s' received on subscription for '%s'", namespace, subInfo.namespace)
-			return nil
-		}
-	} else {
-		namespace = subInfo.namespace
-	}
-
-	batch, err := common.BuildBatchPin(ctx, namespace, event, sUUIDs, sBatchHash, sContexts, sPayloadRef)
-	if err != nil {
-		return nil // move on
-	}
-
-	// If there's an error dispatching the event, we must return the error and shutdown
-	return f.callbacks.BatchPinComplete(ctx, batch, verifier)
+	return f.callbacks.BatchPinOrNetworkAction(ctx, nsOrAction, subInfo, location, event, verifier, params)
 }
 
 func (f *Fabric) buildEventLocationString(chaincode string) string {
@@ -417,31 +380,23 @@ func (f *Fabric) handleReceipt(ctx context.Context, reply fftypes.JSONObject) {
 	f.callbacks.OperationUpdate(ctx, f, requestID, updateType, txHash, message, reply)
 }
 
-func (f *Fabric) AddFireflySubscription(ctx context.Context, namespace string, location *fftypes.JSONAny, firstEvent string) (string, error) {
+func (f *Fabric) AddFireflySubscription(ctx context.Context, namespace core.NamespaceRef, location *fftypes.JSONAny, firstEvent string) (string, error) {
 	fabricOnChainLocation, err := parseContractLocation(ctx, location)
 	if err != nil {
 		return "", err
 	}
 
-	sub, subNS, err := f.streams.ensureFireFlySubscription(ctx, namespace, fabricOnChainLocation, firstEvent, f.streamID, batchPinEvent)
+	version, err := f.GetNetworkVersion(ctx, location)
 	if err != nil {
 		return "", err
 	}
 
-	version, err := f.GetNetworkVersion(ctx, location)
+	sub, err := f.streams.ensureFireFlySubscription(ctx, namespace.LocalName, version, fabricOnChainLocation, firstEvent, f.streamID, batchPinEvent)
 	if err != nil {
 		return "", err
 	}
 
-	if version > 1 && subNS == "" {
-		return "", i18n.NewError(ctx, coremsgs.MsgInvalidSubscriptionForNetwork, sub.Name, version)
-	}
-
-	f.subs[sub.ID] = subscriptionInfo{
-		namespace: subNS,
-		channel:   fabricOnChainLocation.Channel,
-		version:   version,
-	}
+	f.subs.AddSubscription(ctx, namespace, version, sub.ID, fabricOnChainLocation.Channel)
 	return sub.ID, nil
 }
 
@@ -449,11 +404,7 @@ func (f *Fabric) RemoveFireflySubscription(ctx context.Context, subID string) {
 	// Don't actually delete the subscription from fabconnect, as this may be called while processing
 	// events from the subscription (and handling that scenario cleanly could be difficult for fabconnect).
 	// TODO: can old subscriptions be somehow cleaned up later?
-	if _, ok := f.subs[subID]; ok {
-		delete(f.subs, subID)
-	} else {
-		log.L(ctx).Debugf("Invalid subscription ID: %s", subID)
-	}
+	f.subs.RemoveSubscription(ctx, subID)
 }
 
 func (f *Fabric) handleMessageBatch(ctx context.Context, messages []interface{}) error {
@@ -474,10 +425,10 @@ func (f *Fabric) handleMessageBatch(ctx context.Context, messages []interface{})
 		logger.Tracef("Message: %+v", msgJSON)
 
 		// Matches one of the active FireFly BatchPin subscriptions
-		if subInfo, ok := f.subs[sub]; ok {
+		if subInfo := f.subs.GetSubscription(sub); subInfo != nil {
 			location, err := encodeContractLocation(ctx, &Location{
 				Chaincode: msgJSON.GetString("chaincodeId"),
-				Channel:   subInfo.channel,
+				Channel:   subInfo.Extra.(string),
 			})
 			if err != nil {
 				done()
@@ -486,7 +437,7 @@ func (f *Fabric) handleMessageBatch(ctx context.Context, messages []interface{})
 
 			switch eventName {
 			case broadcastBatchEventName:
-				if err := f.handleBatchPinEvent(eventCtx, location, &subInfo, msgJSON); err != nil {
+				if err := f.handleBatchPinEvent(eventCtx, location, subInfo, msgJSON); err != nil {
 					done()
 					return err
 				}
@@ -643,7 +594,7 @@ func hexFormatB32(b *fftypes.Bytes32) string {
 	return "0x" + hex.EncodeToString(b[0:32])
 }
 
-func (f *Fabric) SubmitBatchPin(ctx context.Context, nsOpID string, signingKey string, batch *blockchain.BatchPin, location *fftypes.JSONAny) error {
+func (f *Fabric) SubmitBatchPin(ctx context.Context, nsOpID, remoteNamespace, signingKey string, batch *blockchain.BatchPin, location *fftypes.JSONAny) error {
 	fabricOnChainLocation, err := parseContractLocation(ctx, location)
 	if err != nil {
 		return err
@@ -668,7 +619,7 @@ func (f *Fabric) SubmitBatchPin(ctx context.Context, nsOpID string, signingKey s
 	if version == 1 {
 		prefixItems = batchPinPrefixItemsV1
 		pinInput = map[string]interface{}{
-			"namespace":  batch.Namespace,
+			"namespace":  remoteNamespace,
 			"uuids":      hexFormatB32(&uuids),
 			"batchHash":  hexFormatB32(batch.BatchHash),
 			"payloadRef": batch.BatchPayloadRef,
diff --git a/internal/dataexchange/ffdx/ffdx.go b/internal/dataexchange/ffdx/ffdx.go
index 23c6abfd2..e44acd4a6 100644
--- a/internal/dataexchange/ffdx/ffdx.go
+++ b/internal/dataexchange/ffdx/ffdx.go
@@ -56,7 +56,7 @@ type callbacks struct {
 
 func (cb *callbacks) DXEvent(ctx context.Context, namespace string, event dataexchange.DXEvent) {
 	if handler, ok := cb.handlers[namespace]; ok {
-		handler.DXEvent(event)
+		handler.DXEvent(ctx, event)
 	} else {
 		log.L(ctx).Errorf("unknown namespace on event '%s'", event.EventID())
 		event.Ack()
diff --git a/internal/events/batch_pin_complete.go b/internal/events/batch_pin_complete.go
index eb319dbea..966d9fa4e 100644
--- a/internal/events/batch_pin_complete.go
+++ b/internal/events/batch_pin_complete.go
@@ -31,7 +31,7 @@ import (
 //
 // We must block here long enough to get the payload from the sharedstorage, persist the messages in the correct
 // sequence, and also persist all the data.
-func (em *eventManager) BatchPinComplete(batchPin *blockchain.BatchPin, signingKey *core.VerifierRef) error {
+func (em *eventManager) BatchPinComplete(namespace string, batchPin *blockchain.BatchPin, signingKey *core.VerifierRef) error {
 	if em.multiparty == nil {
 		log.L(em.ctx).Errorf("Ignoring batch pin from non-multiparty network!")
 		return nil
@@ -40,11 +40,10 @@ func (em *eventManager) BatchPinComplete(batchPin *blockchain.BatchPin, signingK
 		log.L(em.ctx).Errorf("Invalid BatchPin transaction - ID is nil")
 		return nil // move on
 	}
-	if batchPin.Namespace != em.namespace.RemoteName {
-		log.L(em.ctx).Debugf("Ignoring batch pin from different namespace '%s'", batchPin.Namespace)
+	if namespace != em.namespace.LocalName {
+		log.L(em.ctx).Debugf("Ignoring batch pin from different namespace '%s'", namespace)
 		return nil // move on
 	}
-	batchPin.Namespace = em.namespace.LocalName
 
 	log.L(em.ctx).Infof("-> BatchPinComplete batch=%s txn=%s signingIdentity=%s", batchPin.BatchID, batchPin.Event.ProtocolID, signingKey.Value)
 	defer func() {
@@ -61,7 +60,7 @@ func (em *eventManager) BatchPinComplete(batchPin *blockchain.BatchPin, signingK
 			if err := em.persistBatchTransaction(ctx, batchPin); err != nil {
 				return err
 			}
-			chainEvent := buildBlockchainEvent(batchPin.Namespace, nil, &batchPin.Event, &core.BlockchainTransactionRef{
+			chainEvent := buildBlockchainEvent(em.namespace.LocalName, nil, &batchPin.Event, &core.BlockchainTransactionRef{
 				Type:         core.TransactionTypeBatchPin,
 				ID:           batchPin.TransactionID,
 				BlockchainID: batchPin.Event.BlockchainTXID,
@@ -103,7 +102,7 @@ func (em *eventManager) persistContexts(ctx context.Context, batchPin *blockchai
 	pins := make([]*core.Pin, len(batchPin.Contexts))
 	for idx, hash := range batchPin.Contexts {
 		pins[idx] = &core.Pin{
-			Namespace: batchPin.Namespace,
+			Namespace: em.namespace.LocalName,
 			Masked:    private,
 			Hash:      hash,
 			Batch:     batchPin.BatchID,
diff --git a/internal/events/event_manager.go b/internal/events/event_manager.go
index a35b27075..d49c577cc 100644
--- a/internal/events/event_manager.go
+++ b/internal/events/event_manager.go
@@ -65,7 +65,7 @@ type EventManager interface {
 	WaitStop()
 
 	// Bound blockchain callbacks
-	BatchPinComplete(batch *blockchain.BatchPin, signingKey *core.VerifierRef) error
+	BatchPinComplete(namespace string, batch *blockchain.BatchPin, signingKey *core.VerifierRef) error
 	BlockchainEvent(event *blockchain.EventWithSubscription) error
 	BlockchainNetworkAction(action string, location *fftypes.JSONAny, event *blockchain.Event, signingKey *core.VerifierRef) error
 
diff --git a/internal/multiparty/manager.go b/internal/multiparty/manager.go
index ee35bcf4b..dcc6a05b0 100644
--- a/internal/multiparty/manager.go
+++ b/internal/multiparty/manager.go
@@ -141,16 +141,12 @@ func (mm *multipartyManager) configureContractCommon(ctx context.Context, migrat
 		}
 	}
 
-	subID, err := mm.blockchain.AddFireflySubscription(ctx, mm.namespace.LocalName, location, firstEvent)
+	subID, err := mm.blockchain.AddFireflySubscription(ctx, mm.namespace.Ref(), location, firstEvent)
 	if err == nil {
-		contracts.Active = &core.MultipartyContract{
-			Location:   location,
-			FirstEvent: firstEvent,
-			Info: core.MultipartyContractInfo{
-				Subscription: subID,
-				Version:      version,
-			},
-		}
+		contracts.Active.Location = location
+		contracts.Active.FirstEvent = firstEvent
+		contracts.Active.Info.Subscription = subID
+		contracts.Active.Info.Version = version
 		err = mm.database.UpsertNamespace(ctx, mm.namespace, true)
 	}
 	return err
diff --git a/internal/multiparty/operations.go b/internal/multiparty/operations.go
index 8ce97cb05..5dfe0beb5 100644
--- a/internal/multiparty/operations.go
+++ b/internal/multiparty/operations.go
@@ -109,8 +109,7 @@ func (mm *multipartyManager) RunOperation(ctx context.Context, op *core.Prepared
 	case batchPinData:
 		batch := data.Batch
 		contract := mm.namespace.Contracts.Active
-		return nil, false, mm.blockchain.SubmitBatchPin(ctx, op.NamespacedIDString(), batch.Key, &blockchain.BatchPin{
-			Namespace:       batch.Namespace,
+		return nil, false, mm.blockchain.SubmitBatchPin(ctx, op.NamespacedIDString(), batch.Namespace, batch.Key, &blockchain.BatchPin{
 			TransactionID:   batch.TX.ID,
 			BatchID:         batch.ID,
 			BatchHash:       batch.Hash,
diff --git a/internal/operations/manager.go b/internal/operations/manager.go
index 355e4a15c..25e1bb4cc 100644
--- a/internal/operations/manager.go
+++ b/internal/operations/manager.go
@@ -27,7 +27,6 @@ import (
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
-	"github.com/hyperledger/firefly/pkg/dataexchange"
 )
 
 type OperationHandler interface {
@@ -44,7 +43,6 @@ type Manager interface {
 	RetryOperation(ctx context.Context, opID *fftypes.UUID) (*core.Operation, error)
 	AddOrReuseOperation(ctx context.Context, op *core.Operation) error
 	SubmitOperationUpdate(plugin core.Named, update *OperationUpdate)
-	TransferResult(dx dataexchange.Plugin, event dataexchange.DXEvent)
 	ResolveOperationByID(ctx context.Context, opID *fftypes.UUID, op *core.OperationUpdateDTO) error
 	Start() error
 	WaitStop()
@@ -165,37 +163,6 @@ func (om *operationsManager) RetryOperation(ctx context.Context, opID *fftypes.U
 	return op, err
 }
 
-func (om *operationsManager) TransferResult(dx dataexchange.Plugin, event dataexchange.DXEvent) {
-
-	tr := event.TransferResult()
-
-	log.L(om.ctx).Infof("Transfer result %s=%s error='%s' manifest='%s' info='%s'", tr.TrackingID, tr.Status, tr.Error, tr.Manifest, tr.Info)
-
-	opUpdate := &OperationUpdate{
-		NamespacedOpID: event.NamespacedID(),
-		Status:         tr.Status,
-		VerifyManifest: dx.Capabilities().Manifest,
-		ErrorMessage:   tr.Error,
-		Output:         tr.Info,
-		OnComplete: func() {
-			event.Ack()
-		},
-	}
-
-	// Pass manifest verification code to the background worker, for once it has loaded the operation
-	if opUpdate.VerifyManifest {
-		if tr.Manifest != "" {
-			// For batches DX passes us a manifest to compare.
-			opUpdate.DXManifest = tr.Manifest
-		} else if tr.Hash != "" {
-			// For blobs DX passes us a hash to compare.
-			opUpdate.DXHash = tr.Hash
-		}
-	}
-
-	om.SubmitOperationUpdate(dx, opUpdate)
-}
-
 func (om *operationsManager) writeOperationSuccess(ctx context.Context, opID *fftypes.UUID, outputs fftypes.JSONObject) {
 	emptyString := ""
 	if err := om.database.ResolveOperation(ctx, om.namespace, opID, core.OpStatusSucceeded, &emptyString, outputs); err != nil {
diff --git a/internal/orchestrator/bound_callbacks.go b/internal/orchestrator/bound_callbacks.go
index 387a4500e..3eeaeee3a 100644
--- a/internal/orchestrator/bound_callbacks.go
+++ b/internal/orchestrator/bound_callbacks.go
@@ -17,7 +17,10 @@
 package orchestrator
 
 import (
+	"context"
+
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
+	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/internal/events"
 	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/pkg/core"
@@ -42,10 +45,35 @@ func (bc *boundCallbacks) OperationUpdate(plugin core.Named, nsOpID string, stat
 	})
 }
 
-func (bc *boundCallbacks) DXEvent(event dataexchange.DXEvent) {
+func (bc *boundCallbacks) DXEvent(ctx context.Context, event dataexchange.DXEvent) {
 	switch event.Type() {
 	case dataexchange.DXEventTypeTransferResult:
-		bc.om.TransferResult(bc.dx, event)
+		tr := event.TransferResult()
+		log.L(ctx).Infof("Transfer result %s=%s error='%s' manifest='%s' info='%s'", tr.TrackingID, tr.Status, tr.Error, tr.Manifest, tr.Info)
+
+		opUpdate := &operations.OperationUpdate{
+			NamespacedOpID: event.NamespacedID(),
+			Status:         tr.Status,
+			VerifyManifest: bc.dx.Capabilities().Manifest,
+			ErrorMessage:   tr.Error,
+			Output:         tr.Info,
+			OnComplete: func() {
+				event.Ack()
+			},
+		}
+
+		// Pass manifest verification code to the background worker, for once it has loaded the operation
+		if opUpdate.VerifyManifest {
+			if tr.Manifest != "" {
+				// For batches DX passes us a manifest to compare.
+				opUpdate.DXManifest = tr.Manifest
+			} else if tr.Hash != "" {
+				// For blobs DX passes us a hash to compare.
+				opUpdate.DXHash = tr.Hash
+			}
+		}
+
+		bc.om.SubmitOperationUpdate(bc.dx, opUpdate)
 	default:
 		bc.ei.DXEvent(bc.dx, event)
 	}
diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go
index 5d5e5cd52..fc2997613 100644
--- a/internal/orchestrator/orchestrator.go
+++ b/internal/orchestrator/orchestrator.go
@@ -355,12 +355,12 @@ func (or *orchestrator) initHandlers(ctx context.Context) (err error) {
 	or.plugins.Database.Plugin.SetHandler(or.namespace.LocalName, or)
 
 	if or.plugins.Blockchain.Plugin != nil {
-		or.plugins.Blockchain.Plugin.SetHandler(or.namespace.RemoteName, or.events)
+		or.plugins.Blockchain.Plugin.SetHandler(or.namespace.LocalName, or.events)
 		or.plugins.Blockchain.Plugin.SetOperationHandler(or.namespace.LocalName, &or.bc)
 	}
 
 	if or.plugins.SharedStorage.Plugin != nil {
-		or.plugins.SharedStorage.Plugin.SetHandler(or.namespace.RemoteName, &or.bc)
+		or.plugins.SharedStorage.Plugin.SetHandler(or.namespace.LocalName, &or.bc)
 	}
 
 	if or.plugins.DataExchange.Plugin != nil {
diff --git a/internal/orchestrator/status.go b/internal/orchestrator/status.go
index 87d5c0bef..04cf28191 100644
--- a/internal/orchestrator/status.go
+++ b/internal/orchestrator/status.go
@@ -39,28 +39,36 @@ func (or *orchestrator) getPlugins() core.NamespaceStatusPlugins {
 	}
 
 	blockchainsArray := make([]*core.NamespaceStatusPlugin, 0)
-	blockchainsArray = append(blockchainsArray, &core.NamespaceStatusPlugin{
-		Name:       or.plugins.Blockchain.Name,
-		PluginType: or.plugins.Blockchain.Plugin.Name(),
-	})
+	if or.plugins.Blockchain.Plugin != nil {
+		blockchainsArray = append(blockchainsArray, &core.NamespaceStatusPlugin{
+			Name:       or.plugins.Blockchain.Name,
+			PluginType: or.plugins.Blockchain.Plugin.Name(),
+		})
+	}
 
 	databasesArray := make([]*core.NamespaceStatusPlugin, 0)
-	databasesArray = append(databasesArray, &core.NamespaceStatusPlugin{
-		Name:       or.plugins.Database.Name,
-		PluginType: or.plugins.Database.Plugin.Name(),
-	})
+	if or.plugins.Database.Plugin != nil {
+		databasesArray = append(databasesArray, &core.NamespaceStatusPlugin{
+			Name:       or.plugins.Database.Name,
+			PluginType: or.plugins.Database.Plugin.Name(),
+		})
+	}
 
 	sharedstorageArray := make([]*core.NamespaceStatusPlugin, 0)
-	sharedstorageArray = append(sharedstorageArray, &core.NamespaceStatusPlugin{
-		Name:       or.plugins.SharedStorage.Name,
-		PluginType: or.plugins.SharedStorage.Plugin.Name(),
-	})
+	if or.plugins.SharedStorage.Plugin != nil {
+		sharedstorageArray = append(sharedstorageArray, &core.NamespaceStatusPlugin{
+			Name:       or.plugins.SharedStorage.Name,
+			PluginType: or.plugins.SharedStorage.Plugin.Name(),
+		})
+	}
 
 	dataexchangeArray := make([]*core.NamespaceStatusPlugin, 0)
-	dataexchangeArray = append(dataexchangeArray, &core.NamespaceStatusPlugin{
-		Name:       or.plugins.DataExchange.Name,
-		PluginType: or.plugins.DataExchange.Plugin.Name(),
-	})
+	if or.plugins.DataExchange.Plugin != nil {
+		dataexchangeArray = append(dataexchangeArray, &core.NamespaceStatusPlugin{
+			Name:       or.plugins.DataExchange.Name,
+			PluginType: or.plugins.DataExchange.Plugin.Name(),
+		})
+	}
 
 	return core.NamespaceStatusPlugins{
 		Blockchain:    blockchainsArray,
@@ -92,19 +100,11 @@ func (or *orchestrator) GetNodeUUID(ctx context.Context) (node *fftypes.UUID) {
 
 func (or *orchestrator) GetStatus(ctx context.Context) (status *core.NamespaceStatus, err error) {
 
-	org, err := or.identity.GetMultipartyRootOrg(ctx)
-	if err != nil {
-		log.L(ctx).Warnf("Failed to query local org for status: %s", err)
-	}
-
 	status = &core.NamespaceStatus{
 		Namespace: or.namespace,
-		Node: core.NamespaceStatusNode{
+		Node: &core.NamespaceStatusNode{
 			Name: config.GetString(coreconfig.NodeName),
 		},
-		Org: core.NamespaceStatusOrg{
-			Name: or.config.Multiparty.Org.Name,
-		},
 		Plugins: or.getPlugins(),
 		Multiparty: core.NamespaceStatusMultiparty{
 			Enabled: or.config.Multiparty.Enabled,
@@ -112,35 +112,41 @@ func (or *orchestrator) GetStatus(ctx context.Context) (status *core.NamespaceSt
 	}
 
 	if or.config.Multiparty.Enabled {
+		status.Org = &core.NamespaceStatusOrg{Name: or.config.Multiparty.Org.Name}
 		status.Multiparty.Contracts = or.namespace.Contracts
-	}
-
-	if org != nil {
-		status.Org.Registered = true
-		status.Org.ID = org.ID
-		status.Org.DID = org.DID
 
-		fb := database.VerifierQueryFactory.NewFilter(ctx)
-		verifiers, _, err := or.database().GetVerifiers(ctx, org.Namespace, fb.And(fb.Eq("identity", org.ID)))
+		org, err := or.identity.GetMultipartyRootOrg(ctx)
 		if err != nil {
-			return nil, err
-		}
-		status.Org.Verifiers = make([]*core.VerifierRef, len(verifiers))
-		for i, v := range verifiers {
-			status.Org.Verifiers[i] = &v.VerifierRef
+			log.L(ctx).Warnf("Failed to query local org for status: %s", err)
 		}
 
-		node, _, err := or.identity.CachedIdentityLookupNilOK(ctx, fmt.Sprintf("%s%s", core.FireFlyNodeDIDPrefix, status.Node.Name))
-		if err != nil {
-			return nil, err
-		}
-		if node != nil && !node.Parent.Equals(org.ID) {
-			log.L(ctx).Errorf("Specified node name is in use by another org: %s", err)
-			node = nil
-		}
-		if node != nil {
-			status.Node.Registered = true
-			status.Node.ID = node.ID
+		if org != nil {
+			status.Org.Registered = true
+			status.Org.ID = org.ID
+			status.Org.DID = org.DID
+
+			fb := database.VerifierQueryFactory.NewFilter(ctx)
+			verifiers, _, err := or.database().GetVerifiers(ctx, org.Namespace, fb.And(fb.Eq("identity", org.ID)))
+			if err != nil {
+				return nil, err
+			}
+			status.Org.Verifiers = make([]*core.VerifierRef, len(verifiers))
+			for i, v := range verifiers {
+				status.Org.Verifiers[i] = &v.VerifierRef
+			}
+
+			node, _, err := or.identity.CachedIdentityLookupNilOK(ctx, fmt.Sprintf("%s%s", core.FireFlyNodeDIDPrefix, status.Node.Name))
+			if err != nil {
+				return nil, err
+			}
+			if node != nil && !node.Parent.Equals(org.ID) {
+				log.L(ctx).Errorf("Specified node name is in use by another org: %s", err)
+				node = nil
+			}
+			if node != nil {
+				status.Node.Registered = true
+				status.Node.ID = node.ID
+			}
 		}
 	}
 
diff --git a/mocks/blockchainmocks/callbacks.go b/mocks/blockchainmocks/callbacks.go
index 12f009959..076f81770 100644
--- a/mocks/blockchainmocks/callbacks.go
+++ b/mocks/blockchainmocks/callbacks.go
@@ -16,13 +16,13 @@ type Callbacks struct {
 	mock.Mock
 }
 
-// BatchPinComplete provides a mock function with given fields: batch, signingKey
-func (_m *Callbacks) BatchPinComplete(batch *blockchain.BatchPin, signingKey *core.VerifierRef) error {
-	ret := _m.Called(batch, signingKey)
+// BatchPinComplete provides a mock function with given fields: namespace, batch, signingKey
+func (_m *Callbacks) BatchPinComplete(namespace string, batch *blockchain.BatchPin, signingKey *core.VerifierRef) error {
+	ret := _m.Called(namespace, batch, signingKey)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(*blockchain.BatchPin, *core.VerifierRef) error); ok {
-		r0 = rf(batch, signingKey)
+	if rf, ok := ret.Get(0).(func(string, *blockchain.BatchPin, *core.VerifierRef) error); ok {
+		r0 = rf(namespace, batch, signingKey)
 	} else {
 		r0 = ret.Error(0)
 	}
diff --git a/mocks/blockchainmocks/plugin.go b/mocks/blockchainmocks/plugin.go
index 15169ec76..723f750ab 100644
--- a/mocks/blockchainmocks/plugin.go
+++ b/mocks/blockchainmocks/plugin.go
@@ -37,18 +37,18 @@ func (_m *Plugin) AddContractListener(ctx context.Context, subscription *core.Co
 }
 
 // AddFireflySubscription provides a mock function with given fields: ctx, namespace, location, firstEvent
-func (_m *Plugin) AddFireflySubscription(ctx context.Context, namespace string, location *fftypes.JSONAny, firstEvent string) (string, error) {
+func (_m *Plugin) AddFireflySubscription(ctx context.Context, namespace core.NamespaceRef, location *fftypes.JSONAny, firstEvent string) (string, error) {
 	ret := _m.Called(ctx, namespace, location, firstEvent)
 
 	var r0 string
-	if rf, ok := ret.Get(0).(func(context.Context, string, *fftypes.JSONAny, string) string); ok {
+	if rf, ok := ret.Get(0).(func(context.Context, core.NamespaceRef, *fftypes.JSONAny, string) string); ok {
 		r0 = rf(ctx, namespace, location, firstEvent)
 	} else {
 		r0 = ret.Get(0).(string)
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, string, *fftypes.JSONAny, string) error); ok {
+	if rf, ok := ret.Get(1).(func(context.Context, core.NamespaceRef, *fftypes.JSONAny, string) error); ok {
 		r1 = rf(ctx, namespace, location, firstEvent)
 	} else {
 		r1 = ret.Error(1)
@@ -341,13 +341,13 @@ func (_m *Plugin) Start() error {
 	return r0
 }
 
-// SubmitBatchPin provides a mock function with given fields: ctx, nsOpID, signingKey, batch, location
-func (_m *Plugin) SubmitBatchPin(ctx context.Context, nsOpID string, signingKey string, batch *blockchain.BatchPin, location *fftypes.JSONAny) error {
-	ret := _m.Called(ctx, nsOpID, signingKey, batch, location)
+// SubmitBatchPin provides a mock function with given fields: ctx, nsOpID, remtoeNamespace, signingKey, batch, location
+func (_m *Plugin) SubmitBatchPin(ctx context.Context, nsOpID string, remtoeNamespace string, signingKey string, batch *blockchain.BatchPin, location *fftypes.JSONAny) error {
+	ret := _m.Called(ctx, nsOpID, remtoeNamespace, signingKey, batch, location)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, *blockchain.BatchPin, *fftypes.JSONAny) error); ok {
-		r0 = rf(ctx, nsOpID, signingKey, batch, location)
+	if rf, ok := ret.Get(0).(func(context.Context, string, string, string, *blockchain.BatchPin, *fftypes.JSONAny) error); ok {
+		r0 = rf(ctx, nsOpID, remtoeNamespace, signingKey, batch, location)
 	} else {
 		r0 = ret.Error(0)
 	}
diff --git a/mocks/dataexchangemocks/callbacks.go b/mocks/dataexchangemocks/callbacks.go
index 0d390497a..af463b028 100644
--- a/mocks/dataexchangemocks/callbacks.go
+++ b/mocks/dataexchangemocks/callbacks.go
@@ -3,6 +3,8 @@
 package dataexchangemocks
 
 import (
+	context "context"
+
 	dataexchange "github.com/hyperledger/firefly/pkg/dataexchange"
 	mock "github.com/stretchr/testify/mock"
 )
@@ -12,7 +14,7 @@ type Callbacks struct {
 	mock.Mock
 }
 
-// DXEvent provides a mock function with given fields: event
-func (_m *Callbacks) DXEvent(event dataexchange.DXEvent) {
-	_m.Called(event)
+// DXEvent provides a mock function with given fields: ctx, event
+func (_m *Callbacks) DXEvent(ctx context.Context, event dataexchange.DXEvent) {
+	_m.Called(ctx, event)
 }
diff --git a/mocks/eventmocks/event_manager.go b/mocks/eventmocks/event_manager.go
index 8a5287a9b..fb59c83c5 100644
--- a/mocks/eventmocks/event_manager.go
+++ b/mocks/eventmocks/event_manager.go
@@ -41,13 +41,13 @@ func (_m *EventManager) AddSystemEventListener(ns string, el system.EventListene
 	return r0
 }
 
-// BatchPinComplete provides a mock function with given fields: batch, signingKey
-func (_m *EventManager) BatchPinComplete(batch *blockchain.BatchPin, signingKey *core.VerifierRef) error {
-	ret := _m.Called(batch, signingKey)
+// BatchPinComplete provides a mock function with given fields: namespace, batch, signingKey
+func (_m *EventManager) BatchPinComplete(namespace string, batch *blockchain.BatchPin, signingKey *core.VerifierRef) error {
+	ret := _m.Called(namespace, batch, signingKey)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(*blockchain.BatchPin, *core.VerifierRef) error); ok {
-		r0 = rf(batch, signingKey)
+	if rf, ok := ret.Get(0).(func(string, *blockchain.BatchPin, *core.VerifierRef) error); ok {
+		r0 = rf(namespace, batch, signingKey)
 	} else {
 		r0 = ret.Error(0)
 	}
diff --git a/mocks/operationmocks/manager.go b/mocks/operationmocks/manager.go
index f0642aab4..5029ad825 100644
--- a/mocks/operationmocks/manager.go
+++ b/mocks/operationmocks/manager.go
@@ -5,10 +5,8 @@ package operationmocks
 import (
 	context "context"
 
-	core "github.com/hyperledger/firefly/pkg/core"
-	dataexchange "github.com/hyperledger/firefly/pkg/dataexchange"
-
 	fftypes "github.com/hyperledger/firefly-common/pkg/fftypes"
+	core "github.com/hyperledger/firefly/pkg/core"
 
 	mock "github.com/stretchr/testify/mock"
 
@@ -148,11 +146,6 @@ func (_m *Manager) SubmitOperationUpdate(plugin core.Named, update *operations.O
 	_m.Called(plugin, update)
 }
 
-// TransferResult provides a mock function with given fields: dx, event
-func (_m *Manager) TransferResult(dx dataexchange.Plugin, event dataexchange.DXEvent) {
-	_m.Called(dx, event)
-}
-
 // WaitStop provides a mock function with given fields:
 func (_m *Manager) WaitStop() {
 	_m.Called()
diff --git a/pkg/blockchain/plugin.go b/pkg/blockchain/plugin.go
index 895e8288a..7a897b1ef 100644
--- a/pkg/blockchain/plugin.go
+++ b/pkg/blockchain/plugin.go
@@ -58,13 +58,13 @@ type Plugin interface {
 	NormalizeSigningKey(ctx context.Context, keyRef string) (string, error)
 
 	// SubmitBatchPin sequences a batch of message globally to all viewers of a given ledger
-	SubmitBatchPin(ctx context.Context, nsOpID string, signingKey string, batch *BatchPin, location *fftypes.JSONAny) error
+	SubmitBatchPin(ctx context.Context, nsOpID, remtoeNamespace, signingKey string, batch *BatchPin, location *fftypes.JSONAny) error
 
 	// SubmitNetworkAction writes a special "BatchPin" event which signals the plugin to take an action
-	SubmitNetworkAction(ctx context.Context, nsOpID string, signingKey string, action core.NetworkActionType, location *fftypes.JSONAny) error
+	SubmitNetworkAction(ctx context.Context, nsOpID, signingKey string, action core.NetworkActionType, location *fftypes.JSONAny) error
 
 	// InvokeContract submits a new transaction to be executed by custom on-chain logic
-	InvokeContract(ctx context.Context, nsOpID string, signingKey string, location *fftypes.JSONAny, method *fftypes.FFIMethod, input map[string]interface{}, options map[string]interface{}) error
+	InvokeContract(ctx context.Context, nsOpID, signingKey string, location *fftypes.JSONAny, method *fftypes.FFIMethod, input map[string]interface{}, options map[string]interface{}) error
 
 	// QueryContract executes a method via custom on-chain logic and returns the result
 	QueryContract(ctx context.Context, location *fftypes.JSONAny, method *fftypes.FFIMethod, input map[string]interface{}, options map[string]interface{}) (interface{}, error)
@@ -94,7 +94,7 @@ type Plugin interface {
 	GetAndConvertDeprecatedContractConfig(ctx context.Context) (location *fftypes.JSONAny, fromBlock string, err error)
 
 	// AddFireflySubscription creates a FireFly BatchPin subscription for the provided location
-	AddFireflySubscription(ctx context.Context, namespace string, location *fftypes.JSONAny, firstEvent string) (subID string, err error)
+	AddFireflySubscription(ctx context.Context, namespace core.NamespaceRef, location *fftypes.JSONAny, firstEvent string) (subID string, err error)
 
 	// RemoveFireFlySubscription removes the provided FireFly subscription
 	RemoveFireflySubscription(ctx context.Context, subID string)
@@ -112,7 +112,7 @@ type Callbacks interface {
 	// submitted by us, or by any other authorized party in the network.
 	//
 	// Error should only be returned in shutdown scenarios
-	BatchPinComplete(batch *BatchPin, signingKey *core.VerifierRef) error
+	BatchPinComplete(namespace string, batch *BatchPin, signingKey *core.VerifierRef) error
 
 	// BlockchainNetworkAction notifies on the arrival of a network operator action
 	//
@@ -131,9 +131,6 @@ type Capabilities struct {
 // BatchPin is the set of data pinned to the blockchain for a batch - whether it's private or broadcast.
 type BatchPin struct {
 
-	// Namespace goes in the clear on the chain (for network rules V1 only)
-	Namespace string
-
 	// TransactionID is the firefly transaction ID allocated before transaction submission for correlation with events (it's a UUID so no leakage)
 	TransactionID *fftypes.UUID
 
diff --git a/pkg/core/namespace_status.go b/pkg/core/namespace_status.go
index 9093a38d0..de35fb745 100644
--- a/pkg/core/namespace_status.go
+++ b/pkg/core/namespace_status.go
@@ -21,8 +21,8 @@ import "github.com/hyperledger/firefly-common/pkg/fftypes"
 // NamespaceStatus is a set of information that represents the configuration and status of a given namespace
 type NamespaceStatus struct {
 	Namespace  *Namespace                `ffstruct:"NamespaceStatus" json:"namespace"`
-	Node       NamespaceStatusNode       `ffstruct:"NamespaceStatus" json:"node"`
-	Org        NamespaceStatusOrg        `ffstruct:"NamespaceStatus" json:"org"`
+	Node       *NamespaceStatusNode      `ffstruct:"NamespaceStatus" json:"node"`
+	Org        *NamespaceStatusOrg       `ffstruct:"NamespaceStatus" json:"org,omitempty"`
 	Plugins    NamespaceStatusPlugins    `ffstruct:"NamespaceStatus" json:"plugins"`
 	Multiparty NamespaceStatusMultiparty `ffstruct:"NamespaceStatus" json:"multiparty"`
 }
diff --git a/pkg/dataexchange/plugin.go b/pkg/dataexchange/plugin.go
index 097d4776c..6f8a22fab 100644
--- a/pkg/dataexchange/plugin.go
+++ b/pkg/dataexchange/plugin.go
@@ -102,7 +102,7 @@ type Plugin interface {
 // Callbacks is the interface provided to the data exchange plugin, to allow it to pass events back to firefly.
 type Callbacks interface {
 	// Event has sub-types as defined below, and can be processed and ack'd asynchronously
-	DXEvent(event DXEvent)
+	DXEvent(ctx context.Context, event DXEvent)
 }
 
 type DXEventType int


Test patch:
diff --git a/internal/blockchain/common/common_test.go b/internal/blockchain/common/common_test.go
new file mode 100644
index 000000000..92e974ec9
--- /dev/null
+++ b/internal/blockchain/common/common_test.go
@@ -0,0 +1,221 @@
+// Copyright © 2022 Kaleido, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package common
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/hyperledger/firefly-common/pkg/fftypes"
+	"github.com/hyperledger/firefly/mocks/blockchainmocks"
+	"github.com/hyperledger/firefly/mocks/coremocks"
+	"github.com/hyperledger/firefly/pkg/blockchain"
+	"github.com/hyperledger/firefly/pkg/core"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+func TestCallbackOperationUpdate(t *testing.T) {
+	nsOpID := "ns1:" + fftypes.NewUUID().String()
+
+	mbi := &blockchainmocks.Plugin{}
+	mcb := &coremocks.OperationCallbacks{}
+	cb := NewBlockchainCallbacks()
+	cb.SetOperationalHandler("ns1", mcb)
+
+	mcb.On("OperationUpdate", mbi, nsOpID, core.OpStatusSucceeded, "tx1", "err", mock.Anything).Return().Once()
+	cb.OperationUpdate(context.Background(), mbi, nsOpID, core.OpStatusSucceeded, "tx1", "err", fftypes.JSONObject{})
+
+	nsOpID = "ns2:" + fftypes.NewUUID().String()
+	cb.OperationUpdate(context.Background(), mbi, nsOpID, core.OpStatusSucceeded, "tx1", "err", fftypes.JSONObject{})
+
+	mcb.AssertExpectations(t)
+}
+
+func TestCallbackBlockchainEvent(t *testing.T) {
+	event := &blockchain.EventWithSubscription{}
+
+	mcb := &blockchainmocks.Callbacks{}
+	cb := NewBlockchainCallbacks()
+	cb.SetHandler("ns1", mcb)
+
+	mcb.On("BlockchainEvent", event).Return(nil).Once()
+	err := cb.BlockchainEvent(context.Background(), "ns1", event)
+	assert.NoError(t, err)
+
+	err = cb.BlockchainEvent(context.Background(), "ns2", event)
+	assert.NoError(t, err)
+
+	mcb.On("BlockchainEvent", event).Return(fmt.Errorf("pop")).Once()
+	err = cb.BlockchainEvent(context.Background(), "", event)
+	assert.EqualError(t, err, "pop")
+
+	mcb.AssertExpectations(t)
+}
+
+func TestCallbackBatchPinBadBatch(t *testing.T) {
+	event := &blockchain.Event{}
+	verifier := &core.VerifierRef{}
+	params := &BatchPinParams{}
+
+	mcb := &blockchainmocks.Callbacks{}
+	cb := NewBlockchainCallbacks()
+	cb.SetHandler("ns1", mcb)
+
+	sub := &SubscriptionInfo{
+		Version:     2,
+		V2Namespace: "ns1",
+	}
+	err := cb.BatchPinOrNetworkAction(context.Background(), "ns1", sub, fftypes.JSONAnyPtr("{}"), event, verifier, params)
+	assert.NoError(t, err)
+
+	mcb.AssertExpectations(t)
+}
+
+func TestCallbackBatchPin(t *testing.T) {
+	event := &blockchain.Event{}
+	verifier := &core.VerifierRef{}
+	params := &BatchPinParams{
+		UUIDs:     "0xe19af8b390604051812d7597d19adfb9847d3bfd074249efb65d3fed15f5b0a6",
+		BatchHash: "0xd71eb138d74c229a388eb0e1abc03f4c7cbb21d4fc4b839fbf0ec73e4263f6be",
+		Contexts: []string{
+			"0x68e4da79f805bca5b912bcda9c63d03e6e867108dabb9b944109aea541ef522a",
+			"0x19b82093de5ce92a01e333048e877e2374354bf846dd034864ef6ffbd6438771",
+		},
+		PayloadRef: "Qmf412jQZiuVUtdgnB36FXFX7xg5V6KEbSJ4dpQuhkLyfD",
+	}
+
+	mcb := &blockchainmocks.Callbacks{}
+	cb := NewBlockchainCallbacks()
+	cb.SetHandler("ns1", mcb)
+
+	sub := &SubscriptionInfo{
+		Version:     2,
+		V2Namespace: "ns1",
+	}
+	mcb.On("BatchPinComplete", "ns1", mock.Anything, verifier).Return(nil).Once()
+	err := cb.BatchPinOrNetworkAction(context.Background(), "", sub, fftypes.JSONAnyPtr("{}"), event, verifier, params)
+	assert.NoError(t, err)
+
+	mcb.On("BatchPinComplete", "ns1", mock.Anything, verifier).Return(fmt.Errorf("pop")).Once()
+	err = cb.BatchPinOrNetworkAction(context.Background(), "", sub, fftypes.JSONAnyPtr("{}"), event, verifier, params)
+	assert.EqualError(t, err, "pop")
+
+	sub = &SubscriptionInfo{
+		Version:     1,
+		V1Namespace: map[string][]string{"ns2": {"ns1", "ns"}},
+	}
+	mcb.On("BatchPinComplete", "ns1", mock.Anything, verifier).Return(nil).Once()
+	err = cb.BatchPinOrNetworkAction(context.Background(), "ns2", sub, fftypes.JSONAnyPtr("{}"), event, verifier, params)
+	assert.NoError(t, err)
+
+	err = cb.BatchPinOrNetworkAction(context.Background(), "ns3", sub, fftypes.JSONAnyPtr("{}"), event, verifier, params)
+	assert.NoError(t, err)
+
+	mcb.AssertExpectations(t)
+}
+
+func TestCallbackNetworkAction(t *testing.T) {
+	event := &blockchain.Event{}
+	verifier := &core.VerifierRef{}
+	params := &BatchPinParams{}
+
+	mcb := &blockchainmocks.Callbacks{}
+	cb := NewBlockchainCallbacks()
+	cb.SetHandler("ns1", mcb)
+
+	sub := &SubscriptionInfo{
+		Version:     2,
+		V2Namespace: "ns1",
+	}
+	mcb.On("BlockchainNetworkAction", "terminate", mock.Anything, mock.Anything, verifier).Return(nil).Once()
+	err := cb.BatchPinOrNetworkAction(context.Background(), "firefly:terminate", sub, fftypes.JSONAnyPtr("{}"), event, verifier, params)
+	assert.NoError(t, err)
+
+	mcb.On("BlockchainNetworkAction", "terminate", mock.Anything, mock.Anything, verifier).Return(fmt.Errorf("pop")).Once()
+	err = cb.BatchPinOrNetworkAction(context.Background(), "firefly:terminate", sub, fftypes.JSONAnyPtr("{}"), event, verifier, params)
+	assert.EqualError(t, err, "pop")
+
+	sub = &SubscriptionInfo{
+		Version:     1,
+		V1Namespace: map[string][]string{"ns2": {"ns1", "ns"}},
+	}
+	mcb.On("BlockchainNetworkAction", "terminate", mock.Anything, mock.Anything, verifier).Return(nil).Once()
+	err = cb.BatchPinOrNetworkAction(context.Background(), "firefly:terminate", sub, fftypes.JSONAnyPtr("{}"), event, verifier, params)
+	assert.NoError(t, err)
+
+	mcb.AssertExpectations(t)
+}
+
+func TestBuildBatchPinErrors(t *testing.T) {
+	event := &blockchain.Event{}
+	params := &BatchPinParams{}
+
+	_, err := buildBatchPin(context.Background(), event, params)
+	assert.Regexp(t, "missing data", err)
+
+	params.BatchHash = "0xd71eb138d74c229a388eb0e1abc03f4c7cbb21d4fc4b839fbf0ec73e4263f6be"
+	params.UUIDs = "BAD"
+	_, err = buildBatchPin(context.Background(), event, params)
+	assert.Regexp(t, "bad uuids", err)
+
+	params.BatchHash = "BAD"
+	params.UUIDs = "0xe19af8b390604051812d7597d19adfb9847d3bfd074249efb65d3fed15f5b0a6"
+	_, err = buildBatchPin(context.Background(), event, params)
+	assert.Regexp(t, "odd length hex string", err)
+
+	params.BatchHash = "0xd71eb138d74c229a388eb0e1abc03f4c7cbb21d4fc4b839fbf0ec73e4263f6be"
+	params.Contexts = append(params.Contexts, "BAD")
+	_, err = buildBatchPin(context.Background(), event, params)
+	assert.Regexp(t, "odd length hex string", err)
+}
+
+func TestGetNamespaceFromSubName(t *testing.T) {
+	ns := GetNamespaceFromSubName("ff-sub-ns1-123")
+	assert.Equal(t, "ns1", ns)
+
+	ns = GetNamespaceFromSubName("BAD")
+	assert.Equal(t, "", ns)
+}
+
+func TestSubscriptionsAddRemoveSubscription(t *testing.T) {
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+
+	subs := NewFireflySubscriptions()
+
+	subs.AddSubscription(context.Background(), ns, 2, "sub1", nil)
+	assert.NotNil(t, subs.GetSubscription("sub1"))
+	subs.RemoveSubscription(context.Background(), "sub1")
+	assert.Nil(t, subs.GetSubscription("sub1"))
+}
+
+func TestSubscriptionsAddRemoveSubscriptionV1(t *testing.T) {
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+
+	subs := NewFireflySubscriptions()
+
+	subs.AddSubscription(context.Background(), ns, 1, "sub1", nil)
+	assert.NotNil(t, subs.GetSubscription("sub1"))
+	subs.RemoveSubscription(context.Background(), "sub1")
+	assert.Nil(t, subs.GetSubscription("sub1"))
+}
+
+func TestSubscriptionsRemoveInvalid(t *testing.T) {
+	subs := NewFireflySubscriptions()
+	subs.RemoveSubscription(context.Background(), "sub1")
+}
diff --git a/internal/blockchain/ethereum/ethereum_test.go b/internal/blockchain/ethereum/ethereum_test.go
index 7f303d55d..7a4676363 100644
--- a/internal/blockchain/ethereum/ethereum_test.go
+++ b/internal/blockchain/ethereum/ethereum_test.go
@@ -93,8 +93,9 @@ func newTestEthereum() (*Ethereum, func()) {
 		wsconn:      wsm,
 		metrics:     mm,
 		cache:       ccache.New(ccache.Configure().MaxSize(100)),
+		callbacks:   common.NewBlockchainCallbacks(),
+		subs:        common.NewFireflySubscriptions(),
 	}
-	e.callbacks = common.NewBlockchainCallbacks()
 	return e, func() {
 		cancel()
 		if e.closed != nil {
@@ -477,6 +478,46 @@ func TestInitAllExistingStreams(t *testing.T) {
 		}))
 	httpmock.RegisterResponder("PATCH", "http://localhost:12345/eventstreams/es12345",
 		httpmock.NewJsonResponderOrPanic(200, &eventStream{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}))
+	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 2))
+	httpmock.RegisterResponder("POST", "http://localhost:12345/subscriptions",
+		httpmock.NewJsonResponderOrPanic(200, subscription{}))
+
+	resetConf(e)
+	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
+	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
+	utEthconnectConf.Set(EthconnectConfigInstanceDeprecated, "0x71C7656EC7ab88b098defB751B7401B5f6d8976F")
+	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
+
+	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
+		"address": "0x71C7656EC7ab88b098defB751B7401B5f6d8976F",
+	}.String())
+
+	err := e.Init(e.ctx, utConfig, e.metrics)
+	assert.NoError(t, err)
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	assert.NoError(t, err)
+
+	assert.Equal(t, 4, httpmock.GetTotalCallCount())
+	assert.Equal(t, "es12345", e.streamID)
+}
+
+func TestInitAllExistingStreamsV1(t *testing.T) {
+	e, cancel := newTestEthereum()
+	defer cancel()
+
+	mockedClient := &http.Client{}
+	httpmock.ActivateNonDefault(mockedClient)
+	defer httpmock.DeactivateAndReset()
+
+	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}}))
+	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
+		httpmock.NewJsonResponderOrPanic(200, []subscription{
+			{ID: "sub12345", Stream: "es12345", Name: "BatchPin_3078373143373635" /* this is the subname for our combo of instance path and BatchPin */},
+		}))
+	httpmock.RegisterResponder("PATCH", "http://localhost:12345/eventstreams/es12345",
+		httpmock.NewJsonResponderOrPanic(200, &eventStream{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 1))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, subscription{}))
@@ -493,7 +534,8 @@ func TestInitAllExistingStreams(t *testing.T) {
 
 	err := e.Init(e.ctx, utConfig, e.metrics)
 	assert.NoError(t, err)
-	_, err = e.AddFireflySubscription(e.ctx, "ns1", location, "oldest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.NoError(t, err)
 
 	assert.Equal(t, 4, httpmock.GetTotalCallCount())
@@ -512,7 +554,7 @@ func TestInitAllExistingStreamsOld(t *testing.T) {
 		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}}))
 	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, []subscription{
-			{ID: "sub12345", Stream: "es12345", Name: "BatchPin_3078373143373635" /* this is the subname for our combo of instance path and BatchPin */},
+			{ID: "sub12345", Stream: "es12345", Name: "BatchPin"},
 		}))
 	httpmock.RegisterResponder("PATCH", "http://localhost:12345/eventstreams/es12345",
 		httpmock.NewJsonResponderOrPanic(200, &eventStream{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}))
@@ -532,7 +574,8 @@ func TestInitAllExistingStreamsOld(t *testing.T) {
 
 	err := e.Init(e.ctx, utConfig, e.metrics)
 	assert.NoError(t, err)
-	_, err = e.AddFireflySubscription(e.ctx, "ns1", location, "oldest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.NoError(t, err)
 
 	assert.Equal(t, 4, httpmock.GetTotalCallCount())
@@ -571,7 +614,8 @@ func TestInitAllExistingStreamsInvalidName(t *testing.T) {
 
 	err := e.Init(e.ctx, utConfig, e.metrics)
 	assert.NoError(t, err)
-	_, err = e.AddFireflySubscription(e.ctx, "ns1", location, "oldest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.Regexp(t, "FF10416", err)
 }
 
@@ -670,7 +714,7 @@ func TestSubmitBatchPinOK(t *testing.T) {
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
 		"address": "0x123",
 	}.String())
-	err := e.SubmitBatchPin(context.Background(), "", addr, batch, location)
+	err := e.SubmitBatchPin(context.Background(), "", "ns1", addr, batch, location)
 
 	assert.NoError(t, err)
 }
@@ -714,7 +758,7 @@ func TestSubmitBatchPinV1(t *testing.T) {
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
 		"address": "0x123",
 	}.String())
-	err := e.SubmitBatchPin(context.Background(), "", addr, batch, location)
+	err := e.SubmitBatchPin(context.Background(), "", "ns1", addr, batch, location)
 
 	assert.NoError(t, err)
 }
@@ -740,7 +784,7 @@ func TestSubmitBatchPinBadLocation(t *testing.T) {
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
 		"bad": "bad",
 	}.String())
-	err := e.SubmitBatchPin(context.Background(), "", addr, batch, location)
+	err := e.SubmitBatchPin(context.Background(), "", "ns1", addr, batch, location)
 
 	assert.Regexp(t, "FF10310", err)
 }
@@ -785,7 +829,7 @@ func TestSubmitBatchEmptyPayloadRef(t *testing.T) {
 		"address": "0x123",
 	}.String())
 
-	err := e.SubmitBatchPin(context.Background(), "", addr, batch, location)
+	err := e.SubmitBatchPin(context.Background(), "", "ns1", addr, batch, location)
 
 	assert.NoError(t, err)
 
@@ -816,7 +860,7 @@ func TestSubmitBatchPinVersionFail(t *testing.T) {
 	httpmock.RegisterResponder("POST", `http://localhost:12345/`,
 		httpmock.NewStringResponder(500, "pop"))
 
-	err := e.SubmitBatchPin(context.Background(), "", addr, batch, location)
+	err := e.SubmitBatchPin(context.Background(), "", "ns1", addr, batch, location)
 
 	assert.Regexp(t, "FF10111.*pop", err)
 
@@ -853,7 +897,7 @@ func TestSubmitBatchPinFail(t *testing.T) {
 			return httpmock.NewStringResponder(500, "pop")(req)
 		})
 
-	err := e.SubmitBatchPin(context.Background(), "", addr, batch, location)
+	err := e.SubmitBatchPin(context.Background(), "", "ns1", addr, batch, location)
 
 	assert.Regexp(t, "FF10111.*pop", err)
 
@@ -893,7 +937,7 @@ func TestSubmitBatchPinError(t *testing.T) {
 			})(req)
 		})
 
-	err := e.SubmitBatchPin(context.Background(), "", addr, batch, location)
+	err := e.SubmitBatchPin(context.Background(), "", "ns1", addr, batch, location)
 
 	assert.Regexp(t, "FF10111.*Unknown error", err)
 }
@@ -980,21 +1024,21 @@ func TestHandleMessageBatchPinOK(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
 		callbacks: common.NewBlockchainCallbacks(),
+		subs:      common.NewFireflySubscriptions(),
 	}
 	e.SetHandler("ns1", em)
-
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", nil,
+	)
 
 	expectedSigningKeyRef := &core.VerifierRef{
 		Type:  core.VerifierTypeEthAddress,
 		Value: "0x91d2b4381a4cd5c7c0f27565a7d4b829844c8635",
 	}
 
-	em.On("BatchPinComplete", mock.Anything, expectedSigningKeyRef, mock.Anything).Return(nil)
+	em.On("BatchPinComplete", "ns1", mock.Anything, expectedSigningKeyRef, mock.Anything).Return(nil)
 
 	var events []interface{}
 	err := json.Unmarshal(data.Bytes(), &events)
@@ -1004,13 +1048,12 @@ func TestHandleMessageBatchPinOK(t *testing.T) {
 
 	em.AssertExpectations(t)
 
-	b := em.Calls[0].Arguments[0].(*blockchain.BatchPin)
-	assert.Equal(t, "ns1", b.Namespace)
+	b := em.Calls[0].Arguments[1].(*blockchain.BatchPin)
 	assert.Equal(t, "e19af8b3-9060-4051-812d-7597d19adfb9", b.TransactionID.String())
 	assert.Equal(t, "847d3bfd-0742-49ef-b65d-3fed15f5b0a6", b.BatchID.String())
 	assert.Equal(t, "d71eb138d74c229a388eb0e1abc03f4c7cbb21d4fc4b839fbf0ec73e4263f6be", b.BatchHash.String())
 	assert.Equal(t, "Qmf412jQZiuVUtdgnB36FXFX7xg5V6KEbSJ4dpQuhkLyfD", b.BatchPayloadRef)
-	assert.Equal(t, expectedSigningKeyRef, em.Calls[0].Arguments[1])
+	assert.Equal(t, expectedSigningKeyRef, em.Calls[0].Arguments[2])
 	assert.Len(t, b.Contexts, 2)
 	assert.Equal(t, "68e4da79f805bca5b912bcda9c63d03e6e867108dabb9b944109aea541ef522a", b.Contexts[0].String())
 	assert.Equal(t, "19b82093de5ce92a01e333048e877e2374354bf846dd034864ef6ffbd6438771", b.Contexts[1].String())
@@ -1027,7 +1070,7 @@ func TestHandleMessageBatchPinOK(t *testing.T) {
 	}
 	assert.Equal(t, info1, b.Event.Info)
 
-	b2 := em.Calls[1].Arguments[0].(*blockchain.BatchPin)
+	b2 := em.Calls[1].Arguments[1].(*blockchain.BatchPin)
 	info2 := fftypes.JSONObject{
 		"address":          "0x1C197604587F046FD40684A8f21f4609FB811A7b",
 		"blockNumber":      "38011",
@@ -1042,127 +1085,6 @@ func TestHandleMessageBatchPinOK(t *testing.T) {
 
 }
 
-func TestHandleMessageBatchPinWrongNS(t *testing.T) {
-	data := fftypes.JSONAnyPtr(`
-[
-  {
-		"address": "0x1C197604587F046FD40684A8f21f4609FB811A7b",
-		"blockNumber": "38011",
-		"transactionIndex": "0x0",
-		"transactionHash": "0xc26df2bf1a733e9249372d61eb11bd8662d26c8129df76890b1beb2f6fa72628",
-		"data": {
-			"author": "0X91D2B4381A4CD5C7C0F27565A7D4B829844C8635",
-			"namespace": "ns2",
-			"uuids": "0xe19af8b390604051812d7597d19adfb9847d3bfd074249efb65d3fed15f5b0a6",
-			"batchHash": "0xd71eb138d74c229a388eb0e1abc03f4c7cbb21d4fc4b839fbf0ec73e4263f6be",
-			"payloadRef": "Qmf412jQZiuVUtdgnB36FXFX7xg5V6KEbSJ4dpQuhkLyfD",
-			"contexts": [
-				"0x68e4da79f805bca5b912bcda9c63d03e6e867108dabb9b944109aea541ef522a",
-				"0x19b82093de5ce92a01e333048e877e2374354bf846dd034864ef6ffbd6438771"
-			]
-    },
-		"subId": "sb-b5b97a4e-a317-4053-6400-1474650efcb5",
-		"signature": "BatchPin(address,uint256,string,bytes32,bytes32,string,bytes32[])",
-		"logIndex": "50",
-		"timestamp": "1620576488"
-  }
-]`)
-
-	e := &Ethereum{
-		callbacks: common.NewBlockchainCallbacks(),
-	}
-
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
-
-	var events []interface{}
-	err := json.Unmarshal(data.Bytes(), &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.NoError(t, err)
-
-}
-
-func TestHandleMessageBatchPinV2(t *testing.T) {
-	data := fftypes.JSONAnyPtr(`
-[
-  {
-		"address": "0x1C197604587F046FD40684A8f21f4609FB811A7b",
-		"blockNumber": "38011",
-		"transactionIndex": "0x0",
-		"transactionHash": "0xc26df2bf1a733e9249372d61eb11bd8662d26c8129df76890b1beb2f6fa72628",
-		"data": {
-			"author": "0X91D2B4381A4CD5C7C0F27565A7D4B829844C8635",
-			"uuids": "0xe19af8b390604051812d7597d19adfb9847d3bfd074249efb65d3fed15f5b0a6",
-			"batchHash": "0xd71eb138d74c229a388eb0e1abc03f4c7cbb21d4fc4b839fbf0ec73e4263f6be",
-			"payloadRef": "Qmf412jQZiuVUtdgnB36FXFX7xg5V6KEbSJ4dpQuhkLyfD",
-			"contexts": [
-				"0x68e4da79f805bca5b912bcda9c63d03e6e867108dabb9b944109aea541ef522a",
-				"0x19b82093de5ce92a01e333048e877e2374354bf846dd034864ef6ffbd6438771"
-			]
-    },
-		"subId": "sb-b5b97a4e-a317-4053-6400-1474650efcb5",
-		"signature": "BatchPin(address,uint256,string,bytes32,bytes32,string,bytes32[])",
-		"logIndex": "50",
-		"timestamp": "1620576488"
-  }
-]`)
-
-	em := &blockchainmocks.Callbacks{}
-	e := &Ethereum{
-		callbacks: common.NewBlockchainCallbacks(),
-	}
-	e.SetHandler("ns1", em)
-
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   2,
-	}
-
-	expectedSigningKeyRef := &core.VerifierRef{
-		Type:  core.VerifierTypeEthAddress,
-		Value: "0x91d2b4381a4cd5c7c0f27565a7d4b829844c8635",
-	}
-
-	em.On("BatchPinComplete", mock.Anything, expectedSigningKeyRef, mock.Anything).Return(nil)
-
-	var events []interface{}
-	err := json.Unmarshal(data.Bytes(), &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.NoError(t, err)
-
-	em.AssertExpectations(t)
-
-	b := em.Calls[0].Arguments[0].(*blockchain.BatchPin)
-	assert.Equal(t, "ns1", b.Namespace)
-	assert.Equal(t, "e19af8b3-9060-4051-812d-7597d19adfb9", b.TransactionID.String())
-	assert.Equal(t, "847d3bfd-0742-49ef-b65d-3fed15f5b0a6", b.BatchID.String())
-	assert.Equal(t, "d71eb138d74c229a388eb0e1abc03f4c7cbb21d4fc4b839fbf0ec73e4263f6be", b.BatchHash.String())
-	assert.Equal(t, "Qmf412jQZiuVUtdgnB36FXFX7xg5V6KEbSJ4dpQuhkLyfD", b.BatchPayloadRef)
-	assert.Equal(t, expectedSigningKeyRef, em.Calls[0].Arguments[1])
-	assert.Len(t, b.Contexts, 2)
-	assert.Equal(t, "68e4da79f805bca5b912bcda9c63d03e6e867108dabb9b944109aea541ef522a", b.Contexts[0].String())
-	assert.Equal(t, "19b82093de5ce92a01e333048e877e2374354bf846dd034864ef6ffbd6438771", b.Contexts[1].String())
-
-	info1 := fftypes.JSONObject{
-		"address":          "0x1C197604587F046FD40684A8f21f4609FB811A7b",
-		"blockNumber":      "38011",
-		"logIndex":         "50",
-		"signature":        "BatchPin(address,uint256,string,bytes32,bytes32,string,bytes32[])",
-		"subId":            "sb-b5b97a4e-a317-4053-6400-1474650efcb5",
-		"transactionHash":  "0xc26df2bf1a733e9249372d61eb11bd8662d26c8129df76890b1beb2f6fa72628",
-		"transactionIndex": "0x0",
-		"timestamp":        "1620576488",
-	}
-	assert.Equal(t, info1, b.Event.Info)
-
-}
-
 func TestHandleMessageBatchPinMissingAddress(t *testing.T) {
 	data := fftypes.JSONAnyPtr(`
 [
@@ -1192,14 +1114,14 @@ func TestHandleMessageBatchPinMissingAddress(t *testing.T) {
 
 	e := &Ethereum{
 		callbacks: common.NewBlockchainCallbacks(),
+		subs:      common.NewFireflySubscriptions(),
 	}
 	e.SetHandler("ns1", em)
-
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", nil,
+	)
 
 	var events []interface{}
 	err := json.Unmarshal(data.Bytes(), &events)
@@ -1209,171 +1131,15 @@ func TestHandleMessageBatchPinMissingAddress(t *testing.T) {
 
 }
 
-func TestHandleMessageBatchPinMissingAuthor(t *testing.T) {
-	data := fftypes.JSONAnyPtr(`
-[
-  {
-		"address": "0x1C197604587F046FD40684A8f21f4609FB811A7b",
-		"blockNumber": "38011",
-		"transactionIndex": "0x0",
-		"transactionHash": "0xc26df2bf1a733e9249372d61eb11bd8662d26c8129df76890b1beb2f6fa72628",
-		"data": {
-			"author": "",
-			"namespace": "ns1",
-			"uuids": "0xe19af8b390604051812d7597d19adfb9847d3bfd074249efb65d3fed15f5b0a6",
-			"batchHash": "0xd71eb138d74c229a388eb0e1abc03f4c7cbb21d4fc4b839fbf0ec73e4263f6be",
-			"payloadRef": "Qmf412jQZiuVUtdgnB36FXFX7xg5V6KEbSJ4dpQuhkLyfD",
-			"contexts": [
-				"0x68e4da79f805bca5b912bcda9c63d03e6e867108dabb9b944109aea541ef522a",
-				"0x19b82093de5ce92a01e333048e877e2374354bf846dd034864ef6ffbd6438771"
-			]
-    },
-		"subId": "sb-b5b97a4e-a317-4053-6400-1474650efcb5",
-		"signature": "BatchPin(address,uint256,string,bytes32,bytes32,string,bytes32[])",
-		"logIndex": "50",
-		"timestamp": "1620576488"
-  }
-]`)
-
-	e := &Ethereum{}
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
-
-	var events []interface{}
-	err := json.Unmarshal(data.Bytes(), &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.NoError(t, err)
-
-}
-
-func TestHandleMessageEmptyPayloadRef(t *testing.T) {
-	data := fftypes.JSONAnyPtr(`
-[
-  {
-		"address": "0x1C197604587F046FD40684A8f21f4609FB811A7b",
-		"blockNumber": "38011",
-		"transactionIndex": "0x0",
-		"transactionHash": "0xc26df2bf1a733e9249372d61eb11bd8662d26c8129df76890b1beb2f6fa72628",
-		"data": {
-			"author": "0X91D2B4381A4CD5C7C0F27565A7D4B829844C8635",
-			"namespace": "ns1",
-			"uuids": "0xe19af8b390604051812d7597d19adfb9847d3bfd074249efb65d3fed15f5b0a6",
-			"batchHash": "0xd71eb138d74c229a388eb0e1abc03f4c7cbb21d4fc4b839fbf0ec73e4263f6be",
-			"payloadRef": "",
-			"contexts": [
-				"0x68e4da79f805bca5b912bcda9c63d03e6e867108dabb9b944109aea541ef522a",
-				"0x19b82093de5ce92a01e333048e877e2374354bf846dd034864ef6ffbd6438771"
-			]
-    },
-		"subId": "sb-b5b97a4e-a317-4053-6400-1474650efcb5",
-		"signature": "BatchPin(address,uint256,string,bytes32,bytes32,string,bytes32[])",
-		"logIndex": "50",
-		"timestamp": "1620576488"
-  }
-]`)
-
-	em := &blockchainmocks.Callbacks{}
-
-	e := &Ethereum{
-		callbacks: common.NewBlockchainCallbacks(),
-	}
-	e.SetHandler("ns1", em)
-
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
-
-	expectedSigningKeyRef := &core.VerifierRef{
-		Type:  core.VerifierTypeEthAddress,
-		Value: "0x91d2b4381a4cd5c7c0f27565a7d4b829844c8635",
-	}
-
-	em.On("BatchPinComplete", mock.Anything, expectedSigningKeyRef, mock.Anything).Return(nil)
-
-	var events []interface{}
-	err := json.Unmarshal(data.Bytes(), &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.NoError(t, err)
-
-	em.AssertExpectations(t)
-
-	b := em.Calls[0].Arguments[0].(*blockchain.BatchPin)
-	assert.Equal(t, "ns1", b.Namespace)
-	assert.Equal(t, "e19af8b3-9060-4051-812d-7597d19adfb9", b.TransactionID.String())
-	assert.Equal(t, "847d3bfd-0742-49ef-b65d-3fed15f5b0a6", b.BatchID.String())
-	assert.Equal(t, "d71eb138d74c229a388eb0e1abc03f4c7cbb21d4fc4b839fbf0ec73e4263f6be", b.BatchHash.String())
-	assert.Empty(t, b.BatchPayloadRef)
-	assert.Equal(t, expectedSigningKeyRef, em.Calls[0].Arguments[1])
-	assert.Len(t, b.Contexts, 2)
-	assert.Equal(t, "68e4da79f805bca5b912bcda9c63d03e6e867108dabb9b944109aea541ef522a", b.Contexts[0].String())
-	assert.Equal(t, "19b82093de5ce92a01e333048e877e2374354bf846dd034864ef6ffbd6438771", b.Contexts[1].String())
-
-}
-
-func TestHandleMessageBatchPinExit(t *testing.T) {
-	data := fftypes.JSONAnyPtr(`
-[
-  {
-		"address": "0x1C197604587F046FD40684A8f21f4609FB811A7b",
-		"blockNumber": "38011",
-		"transactionIndex": "0x1",
-		"transactionHash": "0x0c50dff0893e795293189d9cc5ba0d63c4020d8758ace4a69d02c9d6d43cb695",
-		"data": {
-			"author": "0x91d2b4381a4cd5c7c0f27565a7d4b829844c8635",
-			"namespace": "ns1",
-			"uuids": "0xe19af8b390604051812d7597d19adfb9a04c7cc37d444c2ba3b054e21326697e",
-			"batchHash": "0x9c19a93b6e85fee041f60f097121829e54cd4aa97ed070d1bc76147caf911fed",
-			"payloadRef": "Qmf412jQZiuVUtdgnB36FXFX7xg5V6KEbSJ4dpQuhkLyfD"
-    },
-		"subId": "sb-b5b97a4e-a317-4053-6400-1474650efcb5",
-		"signature": "BatchPin(address,uint256,string,bytes32,bytes32,string,bytes32[])",
-		"logIndex": "51",
-		"timestamp": "1620576488"
-  }
-]`)
-
-	expectedSigningKeyRef := &core.VerifierRef{
-		Type:  core.VerifierTypeEthAddress,
-		Value: "0x91d2b4381a4cd5c7c0f27565a7d4b829844c8635",
-	}
-
-	em := &blockchainmocks.Callbacks{}
-
-	e := &Ethereum{
-		callbacks: common.NewBlockchainCallbacks(),
-	}
-	e.SetHandler("ns1", em)
-
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
-	em.On("BatchPinComplete", mock.Anything, expectedSigningKeyRef, mock.Anything).Return(fmt.Errorf("pop"))
-
-	var events []interface{}
-	err := json.Unmarshal(data.Bytes(), &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.EqualError(t, err, "pop")
-
-	em.AssertExpectations(t)
-}
-
 func TestHandleMessageBatchPinEmpty(t *testing.T) {
-	e := &Ethereum{}
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
+	e := &Ethereum{
+		subs: common.NewFireflySubscriptions(),
 	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", nil,
+	)
 
 	var events []interface{}
 	err := json.Unmarshal([]byte(`
@@ -1390,12 +1156,14 @@ func TestHandleMessageBatchPinEmpty(t *testing.T) {
 }
 
 func TestHandleMessageBatchMissingData(t *testing.T) {
-	e := &Ethereum{}
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
+	e := &Ethereum{
+		subs: common.NewFireflySubscriptions(),
 	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", nil,
+	)
 
 	var events []interface{}
 	err := json.Unmarshal([]byte(`
@@ -1413,12 +1181,15 @@ func TestHandleMessageBatchMissingData(t *testing.T) {
 }
 
 func TestHandleMessageBatchPinBadTransactionID(t *testing.T) {
-	e := &Ethereum{}
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
+	e := &Ethereum{
+		callbacks: common.NewBlockchainCallbacks(),
+		subs:      common.NewFireflySubscriptions(),
 	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", nil,
+	)
 
 	data := fftypes.JSONAnyPtr(`[{
 		"address": "0x1C197604587F046FD40684A8f21f4609FB811A7b",
@@ -1448,12 +1219,14 @@ func TestHandleMessageBatchPinBadTransactionID(t *testing.T) {
 }
 
 func TestHandleMessageBatchPinBadIDentity(t *testing.T) {
-	e := &Ethereum{}
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
+	e := &Ethereum{
+		subs: common.NewFireflySubscriptions(),
 	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", nil,
+	)
 
 	data := fftypes.JSONAnyPtr(`[{
 		"address": "0x1C197604587F046FD40684A8f21f4609FB811A7b",
@@ -1482,76 +1255,6 @@ func TestHandleMessageBatchPinBadIDentity(t *testing.T) {
 	assert.NoError(t, err)
 }
 
-func TestHandleMessageBatchPinBadBatchHash(t *testing.T) {
-	e := &Ethereum{}
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
-
-	data := fftypes.JSONAnyPtr(`[{
-		"address": "0x1C197604587F046FD40684A8f21f4609FB811A7b",
-		"subId": "sb-b5b97a4e-a317-4053-6400-1474650efcb5",
-		"signature": "BatchPin(address,uint256,string,bytes32,bytes32,string,bytes32[])",
-		"blockNumber": "38011",
-		"transactionIndex": "0x1",
-		"transactionHash": "0x0c50dff0893e795293189d9cc5ba0d63c4020d8758ace4a69d02c9d6d43cb695",
-		"data": {
-			"author": "0X91D2B4381A4CD5C7C0F27565A7D4B829844C8635",
-			"namespace": "ns1",
-			"uuids": "0xe19af8b390604051812d7597d19adfb9847d3bfd074249efb65d3fed15f5b0a6",
-			"batchHash": "!good",
-			"payloadRef": "0xeda586bd8f3c4bc1db5c4b5755113b9a9b4174abe28679fdbc219129400dd7ae",
-			"contexts": [
-				"0xb41753f11522d4ef5c4a467972cf54744c04628ff84a1c994f1b288b2f6ec836",
-				"0xc6c683a0fbe15e452e1ecc3751657446e2f645a8231e3ef9f3b4a8eae03c4136"
-			]
-		},
-		"timestamp": "1620576488"
-	}]`)
-	var events []interface{}
-	err := json.Unmarshal(data.Bytes(), &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.NoError(t, err)
-}
-
-func TestHandleMessageBatchPinBadPin(t *testing.T) {
-	e := &Ethereum{}
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
-
-	data := fftypes.JSONAnyPtr(`[{
-		"address": "0x1C197604587F046FD40684A8f21f4609FB811A7b",
-		"subId": "sb-b5b97a4e-a317-4053-6400-1474650efcb5",
-		"signature": "BatchPin(address,uint256,string,bytes32,bytes32,string,bytes32[])",
-		"blockNumber": "38011",
-		"transactionIndex": "0x1",
-		"transactionHash": "0x0c50dff0893e795293189d9cc5ba0d63c4020d8758ace4a69d02c9d6d43cb695",
-		"data": {
-			"author": "0X91D2B4381A4CD5C7C0F27565A7D4B829844C8635",
-			"namespace": "ns1",
-			"uuids": "0xe19af8b390604051812d7597d19adfb9847d3bfd074249efb65d3fed15f5b0a6",
-			"batchHash": "0xd71eb138d74c229a388eb0e1abc03f4c7cbb21d4fc4b839fbf0ec73e4263f6be",
-			"payloadRef": "0xeda586bd8f3c4bc1db5c4b5755113b9a9b4174abe28679fdbc219129400dd7ae",
-			"contexts": [
-				"0xb41753f11522d4ef5c4a467972cf54744c04628ff84a1c994f1b288b2f6ec836",
-				"!good"
-			]
-		},
-		"timestamp": "1620576488"
-	}]`)
-	var events []interface{}
-	err := json.Unmarshal(data.Bytes(), &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.NoError(t, err)
-}
-
 func TestHandleMessageBatchBadJSON(t *testing.T) {
 	e := &Ethereum{}
 	err := e.handleMessageBatch(context.Background(), []interface{}{10, 20})
@@ -1718,11 +1421,6 @@ func TestFormatNil(t *testing.T) {
 	assert.Equal(t, "0x0000000000000000000000000000000000000000000000000000000000000000", ethHexFormatB32(nil))
 }
 
-// func encodeDetails(internalType string) *fftypes.JSONAny {
-// 	result, _ := json.Marshal(&paramDetails{Type: internalType})
-// 	return fftypes.JSONAnyPtrBytes(result)
-// }
-
 func TestAddSubscription(t *testing.T) {
 	e, cancel := newTestEthereum()
 	defer cancel()
@@ -1932,11 +1630,11 @@ func TestHandleMessageContractEventOldSubscription(t *testing.T) {
 		}))
 
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", nil,
+	)
 	e.streams = newTestStreamManager(e.client)
 
 	em.On("BlockchainEvent", mock.MatchedBy(func(e *blockchain.EventWithSubscription) bool {
@@ -2008,11 +1706,11 @@ func TestHandleMessageContractEventErrorOldSubscription(t *testing.T) {
 
 	e.callbacks = common.NewBlockchainCallbacks()
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", nil,
+	)
 	e.streams = newTestStreamManager(e.client)
 	em.On("BlockchainEvent", mock.Anything).Return(fmt.Errorf("pop"))
 
@@ -2071,11 +1769,11 @@ func TestHandleMessageContractEventWithNamespace(t *testing.T) {
 
 	e.callbacks = common.NewBlockchainCallbacks()
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", nil,
+	)
 	e.streams = newTestStreamManager(e.client)
 
 	em.On("BlockchainEvent", mock.MatchedBy(func(e *blockchain.EventWithSubscription) bool {
@@ -2145,11 +1843,11 @@ func TestHandleMessageContractEventNoNamespaceHandlers(t *testing.T) {
 		}))
 
 	e.SetHandler("ns2", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", nil,
+	)
 	e.streams = newTestStreamManager(e.client)
 
 	em.On("BlockchainEvent", mock.MatchedBy(func(e *blockchain.EventWithSubscription) bool {
@@ -2196,11 +1894,11 @@ func TestHandleMessageContractEventSubNameError(t *testing.T) {
 
 	e.callbacks = common.NewBlockchainCallbacks()
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", nil,
+	)
 	e.streams = newTestStreamManager(e.client)
 
 	var events []interface{}
@@ -2244,11 +1942,11 @@ func TestHandleMessageContractEventError(t *testing.T) {
 		}))
 
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", nil,
+	)
 	e.streams = newTestStreamManager(e.client)
 
 	em.On("BlockchainEvent", mock.Anything).Return(fmt.Errorf("pop"))
@@ -2880,64 +2578,14 @@ func TestHandleNetworkAction(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
 		callbacks: common.NewBlockchainCallbacks(),
+		subs:      common.NewFireflySubscriptions(),
 	}
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
-
-	expectedSigningKeyRef := &core.VerifierRef{
-		Type:  core.VerifierTypeEthAddress,
-		Value: "0x91d2b4381a4cd5c7c0f27565a7d4b829844c8635",
-	}
-
-	em.On("BlockchainNetworkAction", "terminate", mock.AnythingOfType("*fftypes.JSONAny"), mock.AnythingOfType("*blockchain.Event"), expectedSigningKeyRef).Return(nil)
-
-	var events []interface{}
-	err := json.Unmarshal(data.Bytes(), &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.NoError(t, err)
-
-	em.AssertExpectations(t)
-
-}
-
-func TestHandleNetworkActionV2(t *testing.T) {
-	data := fftypes.JSONAnyPtr(`
-[
-  {
-		"address": "0x1C197604587F046FD40684A8f21f4609FB811A7b",
-		"blockNumber": "38011",
-		"transactionIndex": "0x0",
-		"transactionHash": "0xc26df2bf1a733e9249372d61eb11bd8662d26c8129df76890b1beb2f6fa72628",
-		"data": {
-			"author": "0X91D2B4381A4CD5C7C0F27565A7D4B829844C8635",
-			"namespace": "firefly:terminate",
-			"uuids": "0x0000000000000000000000000000000000000000000000000000000000000000",
-			"batchHash": "0x0000000000000000000000000000000000000000000000000000000000000000",
-			"payloadRef": "",
-			"contexts": []
-    },
-		"subId": "sb-b5b97a4e-a317-4053-6400-1474650efcb5",
-		"signature": "BatchPin(address,uint256,string,bytes32,bytes32,string,bytes32[])",
-		"logIndex": "50",
-		"timestamp": "1620576488"
-  }
-]`)
-
-	em := &blockchainmocks.Callbacks{}
-	e := &Ethereum{
-		callbacks: common.NewBlockchainCallbacks(),
-	}
-	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   2,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", nil,
+	)
 
 	expectedSigningKeyRef := &core.VerifierRef{
 		Type:  core.VerifierTypeEthAddress,
@@ -2982,13 +2630,14 @@ func TestHandleNetworkActionFail(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Ethereum{
 		callbacks: common.NewBlockchainCallbacks(),
+		subs:      common.NewFireflySubscriptions(),
 	}
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", nil,
+	)
 
 	expectedSigningKeyRef := &core.VerifierRef{
 		Type:  core.VerifierTypeEthAddress,
@@ -3241,7 +2890,8 @@ func TestAddSubBadLocation(t *testing.T) {
 		"bad": "bad",
 	}.String())
 
-	_, err := e.AddFireflySubscription(e.ctx, "ns1", location, "oldest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err := e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.Regexp(t, "FF10310", err)
 }
 
@@ -3261,7 +2911,9 @@ func TestAddAndRemoveFireflySubscription(t *testing.T) {
 	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(200, []subscription{}))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/subscriptions",
-		httpmock.NewJsonResponderOrPanic(200, subscription{}))
+		httpmock.NewJsonResponderOrPanic(200, subscription{
+			ID: "sub1",
+		}))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 2))
 
 	resetConf(e)
@@ -3277,12 +2929,53 @@ func TestAddAndRemoveFireflySubscription(t *testing.T) {
 		"address": "0x123",
 	}.String())
 
-	subID, err := e.AddFireflySubscription(e.ctx, "ns1", location, "newest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	subID, err := e.AddFireflySubscription(e.ctx, ns, location, "newest")
 	assert.NoError(t, err)
-	assert.Len(t, e.subs, 1)
+	assert.NotNil(t, e.subs.GetSubscription("sub1"))
 
 	e.RemoveFireflySubscription(e.ctx, subID)
-	assert.Len(t, e.subs, 0)
+	assert.Nil(t, e.subs.GetSubscription("sub1"))
+}
+
+func TestAddFireflySubscriptionV1(t *testing.T) {
+	e, cancel := newTestEthereum()
+	defer cancel()
+	resetConf(e)
+
+	mockedClient := &http.Client{}
+	httpmock.ActivateNonDefault(mockedClient)
+	defer httpmock.DeactivateAndReset()
+
+	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{}))
+	httpmock.RegisterResponder("POST", "http://localhost:12345/eventstreams",
+		httpmock.NewJsonResponderOrPanic(200, eventStream{ID: "es12345"}))
+	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
+		httpmock.NewJsonResponderOrPanic(200, []subscription{}))
+	httpmock.RegisterResponder("POST", "http://localhost:12345/subscriptions",
+		httpmock.NewJsonResponderOrPanic(200, subscription{
+			ID: "sub1",
+		}))
+	httpmock.RegisterResponder("POST", "http://localhost:12345/", mockNetworkVersion(t, 1))
+
+	resetConf(e)
+	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
+	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
+	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
+	utConfig.AddKnownKey(FireFlyContractConfigKey+".0."+FireFlyContractAddress, "0x71C7656EC7ab88b098defB751B7401B5f6d8976F")
+
+	err := e.Init(e.ctx, utConfig, e.metrics)
+	assert.NoError(t, err)
+
+	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
+		"address": "0x123",
+	}.String())
+
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err = e.AddFireflySubscription(e.ctx, ns, location, "newest")
+	assert.NoError(t, err)
+	assert.NotNil(t, e.subs.GetSubscription("sub1"))
 }
 
 func TestAddFireflySubscriptionQuerySubsFail(t *testing.T) {
@@ -3317,7 +3010,8 @@ func TestAddFireflySubscriptionQuerySubsFail(t *testing.T) {
 		"address": "0x123",
 	}.String())
 
-	_, err = e.AddFireflySubscription(e.ctx, "ns1", location, "oldest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.Regexp(t, "FF10111", err)
 }
 
@@ -3354,7 +3048,8 @@ func TestAddFireflySubscriptionCreateError(t *testing.T) {
 		"address": "0x123",
 	}.String())
 
-	_, err = e.AddFireflySubscription(e.ctx, "ns1", location, "oldest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.Regexp(t, "FF10111", err)
 }
 
@@ -3390,19 +3085,7 @@ func TestAddFireflySubscriptionGetVersionError(t *testing.T) {
 		"address": "0x123",
 	}.String())
 
-	_, err = e.AddFireflySubscription(e.ctx, "ns1", location, "oldest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.Regexp(t, "FF10111", err)
 }
-
-func TestRemoveInvalidSubscription(t *testing.T) {
-	e, _ := newTestEthereum()
-	e.RemoveFireflySubscription(e.ctx, "bad")
-}
-
-func TestCallbacksWrongNamespace(t *testing.T) {
-	e, _ := newTestEthereum()
-	nsOpID := "ns1:" + fftypes.NewUUID().String()
-	e.callbacks.OperationUpdate(context.Background(), e, nsOpID, core.OpStatusSucceeded, "tx123", "", nil)
-	e.callbacks.BatchPinComplete(context.Background(), &blockchain.BatchPin{Namespace: "ns1"}, nil)
-	e.callbacks.BlockchainNetworkAction(context.Background(), "ns1", "terminate", nil, nil, nil)
-}
diff --git a/internal/blockchain/fabric/fabric_test.go b/internal/blockchain/fabric/fabric_test.go
index a7d92698c..ec42f1720 100644
--- a/internal/blockchain/fabric/fabric_test.go
+++ b/internal/blockchain/fabric/fabric_test.go
@@ -67,8 +67,9 @@ func newTestFabric() (*Fabric, func()) {
 		prefixLong:     defaultPrefixLong,
 		wsconn:         wsm,
 		cache:          ccache.New(ccache.Configure().MaxSize(100)),
+		callbacks:      common.NewBlockchainCallbacks(),
+		subs:           common.NewFireflySubscriptions(),
 	}
-	e.callbacks = common.NewBlockchainCallbacks()
 	return e, func() {
 		cancel()
 		if e.closed != nil {
@@ -230,7 +231,7 @@ func TestInitAllExistingStreams(t *testing.T) {
 			{ID: "sub12345", Stream: "es12345", Name: "ns1_BatchPin"},
 		}))
 	httpmock.RegisterResponder("POST", fmt.Sprintf("http://localhost:12345/query"),
-		mockNetworkVersion(t, 1))
+		mockNetworkVersion(t, 2))
 
 	resetConf(e)
 	utFabconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
@@ -246,14 +247,15 @@ func TestInitAllExistingStreams(t *testing.T) {
 
 	err := e.Init(e.ctx, utConfig, &metricsmocks.Manager{})
 	assert.NoError(t, err)
-	_, err = e.AddFireflySubscription(e.ctx, "ns1", location, "oldest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.NoError(t, err)
 
 	assert.Equal(t, 3, httpmock.GetTotalCallCount())
 	assert.Equal(t, "es12345", e.streamID)
 }
 
-func TestInitAllExistingStreamsOld(t *testing.T) {
+func TestInitAllExistingStreamsV1(t *testing.T) {
 	e, cancel := newTestFabric()
 	defer cancel()
 
@@ -284,7 +286,8 @@ func TestInitAllExistingStreamsOld(t *testing.T) {
 
 	err := e.Init(e.ctx, utConfig, &metricsmocks.Manager{})
 	assert.NoError(t, err)
-	_, err = e.AddFireflySubscription(e.ctx, "ns1", location, "oldest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.NoError(t, err)
 
 	assert.Equal(t, 3, httpmock.GetTotalCallCount())
@@ -300,6 +303,8 @@ func TestAddFireflySubscriptionQuerySubsFail(t *testing.T) {
 		httpmock.NewJsonResponderOrPanic(200, []eventStream{{ID: "es12345", WebSocket: eventStreamWebsocket{Topic: "topic1"}}}))
 	httpmock.RegisterResponder("GET", "http://localhost:12345/subscriptions",
 		httpmock.NewJsonResponderOrPanic(500, "pop"))
+	httpmock.RegisterResponder("POST", fmt.Sprintf("http://localhost:12345/query"),
+		mockNetworkVersion(t, 1))
 
 	mockedClient := &http.Client{}
 	httpmock.ActivateNonDefault(mockedClient)
@@ -318,7 +323,8 @@ func TestAddFireflySubscriptionQuerySubsFail(t *testing.T) {
 
 	err := e.Init(e.ctx, utConfig, &metricsmocks.Manager{})
 	assert.NoError(t, err)
-	_, err = e.AddFireflySubscription(e.ctx, "ns1", location, "newest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err = e.AddFireflySubscription(e.ctx, ns, location, "newest")
 	assert.Regexp(t, "pop", err)
 }
 
@@ -353,7 +359,8 @@ func TestAddFireflySubscriptionGetVersionError(t *testing.T) {
 
 	err := e.Init(e.ctx, utConfig, &metricsmocks.Manager{})
 	assert.NoError(t, err)
-	_, err = e.AddFireflySubscription(e.ctx, "ns1", location, "newest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err = e.AddFireflySubscription(e.ctx, ns, location, "newest")
 	assert.Regexp(t, "pop", err)
 }
 
@@ -388,15 +395,16 @@ func TestAddAndRemoveFireflySubscriptionDeprecatedSubName(t *testing.T) {
 
 	err := e.Init(e.ctx, utConfig, &metricsmocks.Manager{})
 	assert.NoError(t, err)
-	subID, err := e.AddFireflySubscription(e.ctx, "ns1", location, "oldest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	subID, err := e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.NoError(t, err)
 
 	assert.Equal(t, 3, httpmock.GetTotalCallCount())
 	assert.Equal(t, "es12345", e.streamID)
-	assert.Len(t, e.subs, 1)
+	assert.NotNil(t, e.subs.GetSubscription(subID))
 
 	e.RemoveFireflySubscription(e.ctx, subID)
-	assert.Len(t, e.subs, 0)
+	assert.Nil(t, e.subs.GetSubscription(subID))
 }
 
 func TestAddFireflySubscriptionInvalidSubName(t *testing.T) {
@@ -430,21 +438,18 @@ func TestAddFireflySubscriptionInvalidSubName(t *testing.T) {
 
 	err := e.Init(e.ctx, utConfig, &metricsmocks.Manager{})
 	assert.NoError(t, err)
-	_, err = e.AddFireflySubscription(e.ctx, "ns1", location, "oldest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.Regexp(t, "FF10416", err)
 }
 
-func TestRemoveUnknownFireflySubscription(t *testing.T) {
-	e, _ := newTestFabric()
-	e.RemoveFireflySubscription(e.ctx, "does not exist")
-}
-
 func TestAddFFSubscriptionBadLocation(t *testing.T) {
 	e, _ := newTestFabric()
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
 		"bad": "bad",
 	}.String())
-	_, err := e.AddFireflySubscription(e.ctx, "ns1", location, "oldest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err := e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.Regexp(t, "F10310", err)
 }
 
@@ -542,6 +547,8 @@ func TestSubQueryCreateError(t *testing.T) {
 		httpmock.NewJsonResponderOrPanic(200, []subscription{}))
 	httpmock.RegisterResponder("POST", "http://localhost:12345/subscriptions",
 		httpmock.NewStringResponder(500, `pop`))
+	httpmock.RegisterResponder("POST", fmt.Sprintf("http://localhost:12345/query"),
+		mockNetworkVersion(t, 1))
 
 	resetConf(e)
 	utFabconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
@@ -558,7 +565,8 @@ func TestSubQueryCreateError(t *testing.T) {
 
 	err := e.Init(e.ctx, utConfig, &metricsmocks.Manager{})
 	assert.NoError(t, err)
-	_, err = e.AddFireflySubscription(e.ctx, "ns1", location, "oldest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.Regexp(t, "FF10284.*pop", err)
 
 }
@@ -598,7 +606,8 @@ func TestSubQueryCreate(t *testing.T) {
 
 	err := e.Init(e.ctx, utConfig, &metricsmocks.Manager{})
 	assert.NoError(t, err)
-	_, err = e.AddFireflySubscription(e.ctx, "ns1", location, "oldest")
+	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
+	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
 	assert.NoError(t, err)
 
 }
@@ -641,7 +650,7 @@ func TestSubmitBatchPinOK(t *testing.T) {
 			return httpmock.NewJsonResponderOrPanic(200, "")(req)
 		})
 
-	err := e.SubmitBatchPin(context.Background(), "", signer, batch, location)
+	err := e.SubmitBatchPin(context.Background(), "", "ns1", signer, batch, location)
 
 	assert.NoError(t, err)
 
@@ -685,7 +694,7 @@ func TestSubmitBatchPinV1(t *testing.T) {
 			return httpmock.NewJsonResponderOrPanic(200, "")(req)
 		})
 
-	err := e.SubmitBatchPin(context.Background(), "", signer, batch, location)
+	err := e.SubmitBatchPin(context.Background(), "", "ns1", signer, batch, location)
 
 	assert.NoError(t, err)
 
@@ -711,7 +720,7 @@ func TestSubmitBatchPinBadLocation(t *testing.T) {
 		"bad": "simplestorage",
 	}.String())
 
-	err := e.SubmitBatchPin(context.Background(), "", signer, batch, location)
+	err := e.SubmitBatchPin(context.Background(), "", "ns1", signer, batch, location)
 
 	assert.Regexp(t, "FF10310", err)
 }
@@ -753,7 +762,7 @@ func TestSubmitBatchEmptyPayloadRef(t *testing.T) {
 			return httpmock.NewJsonResponderOrPanic(200, "")(req)
 		})
 
-	err := e.SubmitBatchPin(context.Background(), "", signer, batch, location)
+	err := e.SubmitBatchPin(context.Background(), "", "ns1", signer, batch, location)
 
 	assert.NoError(t, err)
 
@@ -786,7 +795,7 @@ func TestSubmitBatchPinVersionFail(t *testing.T) {
 	httpmock.RegisterResponder("POST", fmt.Sprintf("http://localhost:12345/query"),
 		httpmock.NewStringResponder(500, "pop"))
 
-	err := e.SubmitBatchPin(context.Background(), "", signer, batch, location)
+	err := e.SubmitBatchPin(context.Background(), "", "ns1", signer, batch, location)
 
 	assert.Regexp(t, "FF10284.*pop", err)
 
@@ -822,7 +831,7 @@ func TestSubmitBatchPinFail(t *testing.T) {
 	httpmock.RegisterResponder("POST", `http://localhost:12345/transactions`,
 		httpmock.NewStringResponder(500, "pop"))
 
-	err := e.SubmitBatchPin(context.Background(), "", signer, batch, location)
+	err := e.SubmitBatchPin(context.Background(), "", "ns1", signer, batch, location)
 
 	assert.Regexp(t, "FF10284.*pop", err)
 
@@ -860,7 +869,7 @@ func TestSubmitBatchPinError(t *testing.T) {
 			"error": "Invalid",
 		}))
 
-	err := e.SubmitBatchPin(context.Background(), "", signer, batch, location)
+	err := e.SubmitBatchPin(context.Background(), "", "ns1", signer, batch, location)
 
 	assert.Regexp(t, "FF10284.*Invalid", err)
 
@@ -984,21 +993,21 @@ func TestHandleMessageBatchPinOK(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
 		callbacks: common.NewBlockchainCallbacks(),
+		subs:      common.NewFireflySubscriptions(),
 	}
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e", "firefly",
+	)
 
 	expectedSigningKeyRef := &core.VerifierRef{
 		Type:  core.VerifierTypeMSPIdentity,
 		Value: "u0vgwu9s00-x509::CN=user2,OU=client::CN=fabric-ca-server",
 	}
 
-	em.On("BatchPinComplete", mock.Anything, expectedSigningKeyRef).Return(nil)
+	em.On("BatchPinComplete", "ns1", mock.Anything, expectedSigningKeyRef).Return(nil)
 
 	var events []interface{}
 	err := json.Unmarshal(data, &events)
@@ -1006,76 +1015,12 @@ func TestHandleMessageBatchPinOK(t *testing.T) {
 	err = e.handleMessageBatch(context.Background(), events)
 	assert.NoError(t, err)
 
-	b := em.Calls[0].Arguments[0].(*blockchain.BatchPin)
-	assert.Equal(t, "ns1", b.Namespace)
-	assert.Equal(t, "e19af8b3-9060-4051-812d-7597d19adfb9", b.TransactionID.String())
-	assert.Equal(t, "847d3bfd-0742-49ef-b65d-3fed15f5b0a6", b.BatchID.String())
-	assert.Equal(t, "d71eb138d74c229a388eb0e1abc03f4c7cbb21d4fc4b839fbf0ec73e4263f6be", b.BatchHash.String())
-	assert.Equal(t, "Qmf412jQZiuVUtdgnB36FXFX7xg5V6KEbSJ4dpQuhkLyfD", b.BatchPayloadRef)
-	assert.Equal(t, expectedSigningKeyRef, em.Calls[1].Arguments[1])
-	assert.Len(t, b.Contexts, 2)
-	assert.Equal(t, "68e4da79f805bca5b912bcda9c63d03e6e867108dabb9b944109aea541ef522a", b.Contexts[0].String())
-	assert.Equal(t, "19b82093de5ce92a01e333048e877e2374354bf846dd034864ef6ffbd6438771", b.Contexts[1].String())
-
-	em.AssertExpectations(t)
-
-}
-
-func TestHandleMessageBatchPinV2(t *testing.T) {
-	payload := base64.StdEncoding.EncodeToString([]byte(fftypes.JSONObject{
-		"signer":     "u0vgwu9s00-x509::CN=user2,OU=client::CN=fabric-ca-server",
-		"timestamp":  fftypes.JSONObject{"seconds": 1630031667, "nanos": 791499000},
-		"namespace":  "",
-		"uuids":      "0xe19af8b390604051812d7597d19adfb9847d3bfd074249efb65d3fed15f5b0a6",
-		"batchHash":  "0xd71eb138d74c229a388eb0e1abc03f4c7cbb21d4fc4b839fbf0ec73e4263f6be",
-		"payloadRef": "Qmf412jQZiuVUtdgnB36FXFX7xg5V6KEbSJ4dpQuhkLyfD",
-		"contexts":   []string{"0x68e4da79f805bca5b912bcda9c63d03e6e867108dabb9b944109aea541ef522a", "0x19b82093de5ce92a01e333048e877e2374354bf846dd034864ef6ffbd6438771"},
-	}.String()))
-	data := []byte(`
-[
-  {
-		"chaincodeId": "firefly",
-		"blockNumber": 91,
-		"transactionId": "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2",
-		"transactionIndex": 2,
-		"eventIndex": 50,
-		"eventName": "BatchPin",
-		"payload": "` + payload + `",
-		"subId": "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
-  }
-]`)
-
-	em := &blockchainmocks.Callbacks{}
-	e := &Fabric{
-		callbacks: common.NewBlockchainCallbacks(),
-	}
-	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   2,
-	}
-
-	expectedSigningKeyRef := &core.VerifierRef{
-		Type:  core.VerifierTypeMSPIdentity,
-		Value: "u0vgwu9s00-x509::CN=user2,OU=client::CN=fabric-ca-server",
-	}
-
-	em.On("BatchPinComplete", mock.Anything, expectedSigningKeyRef).Return(nil)
-
-	var events []interface{}
-	err := json.Unmarshal(data, &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.NoError(t, err)
-
-	b := em.Calls[0].Arguments[0].(*blockchain.BatchPin)
-	assert.Equal(t, "ns1", b.Namespace)
+	b := em.Calls[0].Arguments[1].(*blockchain.BatchPin)
 	assert.Equal(t, "e19af8b3-9060-4051-812d-7597d19adfb9", b.TransactionID.String())
 	assert.Equal(t, "847d3bfd-0742-49ef-b65d-3fed15f5b0a6", b.BatchID.String())
 	assert.Equal(t, "d71eb138d74c229a388eb0e1abc03f4c7cbb21d4fc4b839fbf0ec73e4263f6be", b.BatchHash.String())
 	assert.Equal(t, "Qmf412jQZiuVUtdgnB36FXFX7xg5V6KEbSJ4dpQuhkLyfD", b.BatchPayloadRef)
+	assert.Equal(t, expectedSigningKeyRef, em.Calls[1].Arguments[2])
 	assert.Len(t, b.Contexts, 2)
 	assert.Equal(t, "68e4da79f805bca5b912bcda9c63d03e6e867108dabb9b944109aea541ef522a", b.Contexts[0].String())
 	assert.Equal(t, "19b82093de5ce92a01e333048e877e2374354bf846dd034864ef6ffbd6438771", b.Contexts[1].String())
@@ -1110,14 +1055,14 @@ func TestHandleMessageBatchPinMissingChaincodeID(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
 		callbacks: common.NewBlockchainCallbacks(),
+		subs:      common.NewFireflySubscriptions(),
 	}
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e", "firefly",
+	)
 
 	var events []interface{}
 	err := json.Unmarshal(data, &events)
@@ -1127,131 +1072,6 @@ func TestHandleMessageBatchPinMissingChaincodeID(t *testing.T) {
 
 }
 
-func TestHandleMessageEmptyPayloadRef(t *testing.T) {
-	data := []byte(`
-[
-  {
-		"chaincodeId": "firefly",
-		"blockNumber": 91,
-		"transactionId": "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2",
-		"eventName": "BatchPin",
-		"payload": "eyJzaWduZXIiOiJ1MHZnd3U5czAwLXg1MDk6OkNOPXVzZXIyLE9VPWNsaWVudDo6Q049ZmFicmljLWNhLXNlcnZlciIsInRpbWVzdGFtcCI6eyJzZWNvbmRzIjoxNjMwMDMyMDQwLCJuYW5vcyI6MjI5MjM1MDAwfSwibmFtZXNwYWNlIjoibnMxIiwidXVpZHMiOiIweGUxOWFmOGIzOTA2MDQwNTE4MTJkNzU5N2QxOWFkZmI5ODQ3ZDNiZmQwNzQyNDllZmI2NWQzZmVkMTVmNWIwYTYiLCJiYXRjaEhhc2giOiIweGQ3MWViMTM4ZDc0YzIyOWEzODhlYjBlMWFiYzAzZjRjN2NiYjIxZDRmYzRiODM5ZmJmMGVjNzNlNDI2M2Y2YmUiLCJwYXlsb2FkUmVmIjoiIiwiY29udGV4dHMiOlsiMHg2OGU0ZGE3OWY4MDViY2E1YjkxMmJjZGE5YzYzZDAzZTZlODY3MTA4ZGFiYjliOTQ0MTA5YWVhNTQxZWY1MjJhIiwiMHgxOWI4MjA5M2RlNWNlOTJhMDFlMzMzMDQ4ZTg3N2UyMzc0MzU0YmY4NDZkZDAzNDg2NGVmNmZmYmQ2NDM4NzcxIl19",
-		"subId": "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
-  }
-]`)
-
-	em := &blockchainmocks.Callbacks{}
-	e := &Fabric{
-		callbacks: common.NewBlockchainCallbacks(),
-	}
-	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
-
-	expectedSigningKeyRef := &core.VerifierRef{
-		Type:  core.VerifierTypeMSPIdentity,
-		Value: "u0vgwu9s00-x509::CN=user2,OU=client::CN=fabric-ca-server",
-	}
-
-	em.On("BatchPinComplete", mock.Anything, expectedSigningKeyRef, mock.Anything, mock.Anything).Return(nil)
-
-	var events []interface{}
-	err := json.Unmarshal(data, &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.NoError(t, err)
-
-	b := em.Calls[0].Arguments[0].(*blockchain.BatchPin)
-	assert.Equal(t, "ns1", b.Namespace)
-	assert.Equal(t, "e19af8b3-9060-4051-812d-7597d19adfb9", b.TransactionID.String())
-	assert.Equal(t, "847d3bfd-0742-49ef-b65d-3fed15f5b0a6", b.BatchID.String())
-	assert.Equal(t, "d71eb138d74c229a388eb0e1abc03f4c7cbb21d4fc4b839fbf0ec73e4263f6be", b.BatchHash.String())
-	assert.Empty(t, b.BatchPayloadRef)
-	assert.Equal(t, expectedSigningKeyRef, em.Calls[0].Arguments[1])
-	assert.Len(t, b.Contexts, 2)
-	assert.Equal(t, "68e4da79f805bca5b912bcda9c63d03e6e867108dabb9b944109aea541ef522a", b.Contexts[0].String())
-	assert.Equal(t, "19b82093de5ce92a01e333048e877e2374354bf846dd034864ef6ffbd6438771", b.Contexts[1].String())
-
-	em.AssertExpectations(t)
-
-}
-
-func TestHandleMessageBatchPinExit(t *testing.T) {
-	data := []byte(`
-[
-  {
-		"chaincodeId": "firefly",
-		"blockNumber": 91,
-		"transactionId": "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2",
-		"eventName": "BatchPin",
-		"payload": "eyJzaWduZXIiOiJ1MHZnd3U5czAwLXg1MDk6OkNOPXVzZXIyLE9VPWNsaWVudDo6Q049ZmFicmljLWNhLXNlcnZlciIsInRpbWVzdGFtcCI6eyJzZWNvbmRzIjoxNjMwMDMyMDQwLCJuYW5vcyI6MjI5MjM1MDAwfSwibmFtZXNwYWNlIjoibnMxIiwidXVpZHMiOiIweGUxOWFmOGIzOTA2MDQwNTE4MTJkNzU5N2QxOWFkZmI5ODQ3ZDNiZmQwNzQyNDllZmI2NWQzZmVkMTVmNWIwYTYiLCJiYXRjaEhhc2giOiIweGQ3MWViMTM4ZDc0YzIyOWEzODhlYjBlMWFiYzAzZjRjN2NiYjIxZDRmYzRiODM5ZmJmMGVjNzNlNDI2M2Y2YmUiLCJwYXlsb2FkUmVmIjoiIiwiY29udGV4dHMiOlsiMHg2OGU0ZGE3OWY4MDViY2E1YjkxMmJjZGE5YzYzZDAzZTZlODY3MTA4ZGFiYjliOTQ0MTA5YWVhNTQxZWY1MjJhIiwiMHgxOWI4MjA5M2RlNWNlOTJhMDFlMzMzMDQ4ZTg3N2UyMzc0MzU0YmY4NDZkZDAzNDg2NGVmNmZmYmQ2NDM4NzcxIl19",
-		"subId": "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
-  }
-]`)
-
-	em := &blockchainmocks.Callbacks{}
-	e := &Fabric{
-		callbacks: common.NewBlockchainCallbacks(),
-	}
-	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
-
-	expectedSigningKeyRef := &core.VerifierRef{
-		Type:  core.VerifierTypeMSPIdentity,
-		Value: "u0vgwu9s00-x509::CN=user2,OU=client::CN=fabric-ca-server",
-	}
-
-	em.On("BatchPinComplete", mock.Anything, expectedSigningKeyRef, mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
-
-	var events []interface{}
-	err := json.Unmarshal(data, &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.EqualError(t, err, "pop")
-
-}
-
-func TestHandleMessageBatchPinEmpty(t *testing.T) {
-	data := []byte(`
-[
-  {
-		"chaincodeId": "firefly",
-		"blockNumber": 91,
-		"transactionId": "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2",
-		"eventName": "BatchPin",
-		"subId": "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
-  }
-]`)
-
-	em := &blockchainmocks.Callbacks{}
-	e := &Fabric{
-		callbacks: common.NewBlockchainCallbacks(),
-	}
-	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
-
-	var events []interface{}
-	err := json.Unmarshal(data, &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.NoError(t, err)
-	assert.Equal(t, 0, len(em.Calls))
-}
-
 func TestHandleMessageUnknownEventName(t *testing.T) {
 	data := []byte(`
 [
@@ -1267,73 +1087,15 @@ func TestHandleMessageUnknownEventName(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
 		callbacks: common.NewBlockchainCallbacks(),
+		subs:      common.NewFireflySubscriptions(),
 	}
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
-
-	var events []interface{}
-	err := json.Unmarshal(data, &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.NoError(t, err)
-	assert.Equal(t, 0, len(em.Calls))
-}
-
-func TestHandleMessageBatchPinBadBatchHash(t *testing.T) {
-	em := &blockchainmocks.Callbacks{}
-	e := &Fabric{
-		callbacks: common.NewBlockchainCallbacks(),
-	}
-	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e", "firefly",
+	)
 
-	data := []byte(`[{
-		"chaincodeId": "firefly",
-		"blockNumber": 91,
-		"transactionId": "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2",
-		"eventName": "BatchPin",
-		"payload": "eyJzaWduZXIiOiJ1MHZnd3U5czAwLXg1MDk6OkNOPXVzZXIyLE9VPWNsaWVudDo6Q049ZmFicmljLWNhLXNlcnZlciIsInRpbWVzdGFtcCI6eyJzZWNvbmRzIjoxNjMwMDMzMjMzLCJuYW5vcyI6NTAwMDc3MDAwfSwibmFtZXNwYWNlIjoibnMxIiwidXVpZHMiOiIweGUxOWFmOGIzOTA2MDQwNTE4MTJkNzU5N2QxOWFkZmI5ODQ3ZDNiZmQwNzQyNDllZmI2NWQzZmVkMTVmNWIwYTYiLCJiYXRjaEhhc2giOiIhZ29vZCIsInBheWxvYWRSZWYiOiJRbWY0MTJqUVppdVZVdGRnbkIzNkZYRlg3eGc1VjZLRWJTSjRkcFF1aGtMeWZEIiwiY29udGV4dHMiOltdfQ==",
-		"subId": "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
-	}]`)
-	var events []interface{}
-	err := json.Unmarshal(data, &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.NoError(t, err)
-	assert.Equal(t, 0, len(em.Calls))
-}
-
-func TestHandleMessageBatchPinBadPin(t *testing.T) {
-	em := &blockchainmocks.Callbacks{}
-	e := &Fabric{
-		callbacks: common.NewBlockchainCallbacks(),
-	}
-	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
-
-	data := []byte(`[{
-		"chaincodeId": "firefly",
-		"blockNumber": 91,
-		"transactionId": "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2",
-		"eventName": "BatchPin",
-		"payload": "eyJzaWduZXIiOiJ1MHZnd3U5czAwLXg1MDk6OkNOPXVzZXIyLE9VPWNsaWVudDo6Q049ZmFicmljLWNhLXNlcnZlciIsInRpbWVzdGFtcCI6eyJzZWNvbmRzIjoxNjMwMDMzMzQ0LCJuYW5vcyI6OTY1NjE4MDAwfSwibmFtZXNwYWNlIjoibnMxIiwidXVpZHMiOiIweGUxOWFmOGIzOTA2MDQwNTE4MTJkNzU5N2QxOWFkZmI5ODQ3ZDNiZmQwNzQyNDllZmI2NWQzZmVkMTVmNWIwYTYiLCJiYXRjaEhhc2giOiIweGQ3MWViMTM4ZDc0YzIyOWEzODhlYjBlMWFiYzAzZjRjN2NiYjIxZDRmYzRiODM5ZmJmMGVjNzNlNDI2M2Y2YmUiLCJwYXlsb2FkUmVmIjoiUW1mNDEyalFaaXVWVXRkZ25CMzZGWEZYN3hnNVY2S0ViU0o0ZHBRdWhrTHlmRCIsImNvbnRleHRzIjpbIjB4NjhlNGRhNzlmODA1YmNhNWI5MTJiY2RhOWM2M2QwM2U2ZTg2NzEwOGRhYmI5Yjk0NDEwOWFlYTU0MWVmNTIyYSIsIiFnb29kIl19",
-		"subId": "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
-	}]`)
 	var events []interface{}
 	err := json.Unmarshal(data, &events)
 	assert.NoError(t, err)
@@ -1346,14 +1108,14 @@ func TestHandleMessageBatchPinBadPayloadEncoding(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
 		callbacks: common.NewBlockchainCallbacks(),
+		subs:      common.NewFireflySubscriptions(),
 	}
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e", "firefly",
+	)
 
 	data := []byte(`[{
 		"chaincodeId": "firefly",
@@ -1371,35 +1133,6 @@ func TestHandleMessageBatchPinBadPayloadEncoding(t *testing.T) {
 	assert.Equal(t, 0, len(em.Calls))
 }
 
-func TestHandleMessageBatchPinBadPayloadUUIDs(t *testing.T) {
-	em := &blockchainmocks.Callbacks{}
-	e := &Fabric{
-		callbacks: common.NewBlockchainCallbacks(),
-	}
-	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
-
-	data := []byte(`[{
-		"chaincodeId": "firefly",
-		"blockNumber": 91,
-		"transactionId": "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2",
-		"eventName": "BatchPin",
-		"payload": "eyJzaWduZXIiOiJPcmcxTVNQOjp4NTA5OjpDTj0weDA3YTA5YzE2ZWQ5ZWYyYmIwYmNiYzUxNzk4OGU4MmIzNzA0NDk4YzQsT1U9Y2xpZW50OjpDTj1mYWJyaWNfY2Eub3JnMS5leGFtcGxlLmNvbSxPVT1IeXBlcmxlZGdlciBGYWJyaWMsTz1vcmcxLmV4YW1wbGUuY29tLEw9U2FuIEZyYW5jaXNjbyxTVD1DYWxpZm9ybmlhLEM9VVMiLCJ0aW1lc3RhbXAiOnsic2Vjb25kcyI6MTYzNDMwNDAzNSwibmFub3MiOjI5OTcwMjUwMH0sIm5hbWVzcGFjZSI6ImRlZmF1bHQiLCJ1dWlkcyI6IjB4MjYxNjY2OGExYjIxNGFkY2JkN2IyOGE3ZjkxMDM3MjNiNzEwMTk4ODc4NWE0NzZmYTM2YjM1OWUyZCIsImJhdGNoSGFzaCI6IjB4ZDRkYjliNmQ3YWYzNWQyYTU4ZDgwYmFlY2QxOTI2MjM0Mzg0YmIxODljMGQ2YmRmMzQzNGMyZmE5YzY2MGM0MiIsInBheWxvYWRSZWYiOiJRbWNuRUVjY0tkV0tHZDZqV2ZaNGZOV2dqTnBVSm15bm5lV0tMRW11Rjh3UlNDIiwiY29udGV4dHMiOlsiMHgzN2E4ZWVjMWNlMTk2ODdkMTMyZmUyOTA1MWRjYTYyOWQxNjRlMmM0OTU4YmExNDFkNWY0MTMzYTMzZjA2ODhmIl19",
-		"subId": "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
-	}]`)
-	var events []interface{}
-	err := json.Unmarshal(data, &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.NoError(t, err)
-	assert.Equal(t, 0, len(em.Calls))
-}
-
 func TestHandleMessageBatchBadJSON(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
@@ -1502,6 +1235,7 @@ func TestHandleReceiptTXSuccess(t *testing.T) {
 		ctx:       context.Background(),
 		topic:     "topic1",
 		callbacks: common.NewBlockchainCallbacks(),
+		subs:      common.NewFireflySubscriptions(),
 		wsconn:    wsm,
 	}
 	e.SetOperationHandler("ns1", em)
@@ -1544,6 +1278,7 @@ func TestHandleReceiptNoRequestID(t *testing.T) {
 		ctx:       context.Background(),
 		topic:     "topic1",
 		callbacks: common.NewBlockchainCallbacks(),
+		subs:      common.NewFireflySubscriptions(),
 		wsconn:    wsm,
 	}
 	e.SetHandler("ns1", em)
@@ -1562,6 +1297,7 @@ func TestHandleReceiptFailedTx(t *testing.T) {
 		ctx:       context.Background(),
 		topic:     "topic1",
 		callbacks: common.NewBlockchainCallbacks(),
+		subs:      common.NewFireflySubscriptions(),
 		wsconn:    wsm,
 	}
 	e.SetOperationHandler("ns1", em)
@@ -1744,7 +1480,7 @@ func TestHandleMessageContractEventOldSubscription(t *testing.T) {
 [
 	{
 		"chaincodeId": "basic",
-	  "blockNumber": 10,
+		"blockNumber": 10,
 		"transactionId": "4763a0c50e3bba7cef1a7ba35dd3f9f3426bb04d0156f326e84ec99387c4746d",
 		"transactionIndex": 20,
 		"eventIndex": 30,
@@ -1768,12 +1504,6 @@ func TestHandleMessageContractEventOldSubscription(t *testing.T) {
 	e.streams = newTestStreamManager(e.client, e.signer)
 	e.callbacks = common.NewBlockchainCallbacks()
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
 
 	em.On("BlockchainEvent", mock.MatchedBy(func(e *blockchain.EventWithSubscription) bool {
 		assert.Equal(t, "4763a0c50e3bba7cef1a7ba35dd3f9f3426bb04d0156f326e84ec99387c4746d", e.BlockchainTXID)
@@ -1853,12 +1583,6 @@ func TestHandleMessageContractEventNamespacedHandlers(t *testing.T) {
 	e.streams = newTestStreamManager(e.client, e.signer)
 	e.callbacks = common.NewBlockchainCallbacks()
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
 
 	em.On("BlockchainEvent", mock.MatchedBy(func(e *blockchain.EventWithSubscription) bool {
 		assert.Equal(t, "000000000010/000020/000030", e.Event.ProtocolID)
@@ -1927,12 +1651,6 @@ func TestHandleMessageContractEventNoNamespacedHandlers(t *testing.T) {
 	e.streams = newTestStreamManager(e.client, e.signer)
 	e.callbacks = common.NewBlockchainCallbacks()
 	e.SetHandler("ns2", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
 
 	em.On("BlockchainEvent", mock.MatchedBy(func(e *blockchain.EventWithSubscription) bool {
 		assert.Equal(t, "4763a0c50e3bba7cef1a7ba35dd3f9f3426bb04d0156f326e84ec99387c4746d", e.BlockchainTXID)
@@ -1977,52 +1695,11 @@ func TestHandleMessageContractEventNoPayload(t *testing.T) {
 	e.callbacks = common.NewBlockchainCallbacks()
 	e.SetHandler("ns1", em)
 
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
-
-	var events []interface{}
-	err := json.Unmarshal(data, &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.NoError(t, err)
-}
-
-func TestHandleMessageContractEventBadPayload(t *testing.T) {
-	data := []byte(`
-[
-	{
-		"chaincodeId": "basic",
-	  "blockNumber": 10,
-		"transactionId": "4763a0c50e3bba7cef1a7ba35dd3f9f3426bb04d0156f326e84ec99387c4746d",
-		"eventName": "AssetCreated",
-		"payload": "bad",
-		"subId": "sb-cb37cc07-e873-4f58-44ab-55add6bba320"
-	}
-]`)
-
-	em := &blockchainmocks.Callbacks{}
-	e := &Fabric{
-		callbacks: common.NewBlockchainCallbacks(),
-	}
-	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-cb37cc07-e873-4f58-44ab-55add6bba320"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
-
 	var events []interface{}
 	err := json.Unmarshal(data, &events)
 	assert.NoError(t, err)
 	err = e.handleMessageBatch(context.Background(), events)
 	assert.NoError(t, err)
-
-	em.AssertExpectations(t)
 }
 
 func TestHandleMessageContractOldSubError(t *testing.T) {
@@ -2030,7 +1707,7 @@ func TestHandleMessageContractOldSubError(t *testing.T) {
 [
 	{
 		"chaincodeId": "basic",
-	  "blockNumber": 10,
+		"blockNumber": 10,
 		"transactionId": "4763a0c50e3bba7cef1a7ba35dd3f9f3426bb04d0156f326e84ec99387c4746d",
 		"eventName": "AssetCreated",
 		"payload": "eyJBcHByYWlzZWRWYWx1ZSI6MTAsIkNvbG9yIjoicmVkIiwiSUQiOiIxMjM0IiwiT3duZXIiOiJtZSIsIlNpemUiOjN9",
@@ -2052,12 +1729,11 @@ func TestHandleMessageContractOldSubError(t *testing.T) {
 	e.streams = newTestStreamManager(e.client, e.signer)
 	e.callbacks = common.NewBlockchainCallbacks()
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", "firefly",
+	)
 
 	em.On("BlockchainEvent", mock.Anything).Return(fmt.Errorf("pop"))
 
@@ -2075,7 +1751,7 @@ func TestHandleMessageContractEventError(t *testing.T) {
 [
 	{
 		"chaincodeId": "basic",
-	  "blockNumber": 10,
+		"blockNumber": 10,
 		"transactionId": "4763a0c50e3bba7cef1a7ba35dd3f9f3426bb04d0156f326e84ec99387c4746d",
 		"eventName": "AssetCreated",
 		"payload": "eyJBcHByYWlzZWRWYWx1ZSI6MTAsIkNvbG9yIjoicmVkIiwiSUQiOiIxMjM0IiwiT3duZXIiOiJtZSIsIlNpemUiOjN9",
@@ -2097,12 +1773,6 @@ func TestHandleMessageContractEventError(t *testing.T) {
 	e.streams = newTestStreamManager(e.client, e.signer)
 	e.callbacks = common.NewBlockchainCallbacks()
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
 
 	em.On("BlockchainEvent", mock.Anything).Return(fmt.Errorf("pop"))
 
@@ -2120,7 +1790,7 @@ func TestHandleMessageContractGetSubError(t *testing.T) {
 [
 	{
 		"chaincodeId": "basic",
-	  "blockNumber": 10,
+		"blockNumber": 10,
 		"transactionId": "4763a0c50e3bba7cef1a7ba35dd3f9f3426bb04d0156f326e84ec99387c4746d",
 		"eventName": "AssetCreated",
 		"payload": "eyJBcHByYWlzZWRWYWx1ZSI6MTAsIkNvbG9yIjoicmVkIiwiSUQiOiIxMjM0IiwiT3duZXIiOiJtZSIsIlNpemUiOjN9",
@@ -2140,12 +1810,11 @@ func TestHandleMessageContractGetSubError(t *testing.T) {
 	e.streams = newTestStreamManager(e.client, e.signer)
 	e.callbacks = common.NewBlockchainCallbacks()
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-b5b97a4e-a317-4053-6400-1474650efcb5"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-b5b97a4e-a317-4053-6400-1474650efcb5", "firefly",
+	)
 
 	var events []interface{}
 	err := json.Unmarshal(data, &events)
@@ -2516,60 +2185,14 @@ func TestHandleNetworkAction(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
 		callbacks: common.NewBlockchainCallbacks(),
+		subs:      common.NewFireflySubscriptions(),
 	}
 	e.SetHandler("ns1", em)
-
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
-
-	expectedSigningKeyRef := &core.VerifierRef{
-		Type:  core.VerifierTypeMSPIdentity,
-		Value: "u0vgwu9s00-x509::CN=user2,OU=client::CN=fabric-ca-server",
-	}
-
-	em.On("BlockchainNetworkAction", "terminate", mock.AnythingOfType("*fftypes.JSONAny"), mock.AnythingOfType("*blockchain.Event"), expectedSigningKeyRef).Return(nil)
-
-	var events []interface{}
-	err := json.Unmarshal(data, &events)
-	assert.NoError(t, err)
-	err = e.handleMessageBatch(context.Background(), events)
-	assert.NoError(t, err)
-
-	em.AssertExpectations(t)
-
-}
-
-func TestHandleNetworkActionV2(t *testing.T) {
-	data := []byte(`
-[
-  {
-		"chaincodeId": "firefly",
-		"blockNumber": 91,
-		"transactionId": "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2",
-		"transactionIndex": 2,
-		"eventIndex": 50,
-		"eventName": "BatchPin",
-		"payload": "eyJzaWduZXIiOiJ1MHZnd3U5czAwLXg1MDk6OkNOPXVzZXIyLE9VPWNsaWVudDo6Q049ZmFicmljLWNhLXNlcnZlciIsInRpbWVzdGFtcCI6eyJzZWNvbmRzIjoxNjMwMDMxNjY3LCJuYW5vcyI6NzkxNDk5MDAwfSwibmFtZXNwYWNlIjoiZmlyZWZseTp0ZXJtaW5hdGUiLCJ1dWlkcyI6IjB4MDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMCIsImJhdGNoSGFzaCI6IjB4MDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMCIsInBheWxvYWRSZWYiOiIiLCJjb250ZXh0cyI6W119",
-		"subId": "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"
-  }
-]`)
-
-	em := &blockchainmocks.Callbacks{}
-	e := &Fabric{
-		callbacks: common.NewBlockchainCallbacks(),
-	}
-	e.SetHandler("ns1", em)
-
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   2,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e", "firefly",
+	)
 
 	expectedSigningKeyRef := &core.VerifierRef{
 		Type:  core.VerifierTypeMSPIdentity,
@@ -2606,14 +2229,14 @@ func TestHandleNetworkActionFail(t *testing.T) {
 	em := &blockchainmocks.Callbacks{}
 	e := &Fabric{
 		callbacks: common.NewBlockchainCallbacks(),
+		subs:      common.NewFireflySubscriptions(),
 	}
 	e.SetHandler("ns1", em)
-	e.subs = map[string]subscriptionInfo{}
-	e.subs["sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e"] = subscriptionInfo{
-		namespace: "ns1",
-		channel:   "firefly",
-		version:   1,
-	}
+	e.subs.AddSubscription(
+		context.Background(),
+		core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"},
+		1, "sb-0910f6a8-7bd6-4ced-453e-2db68149ce8e", "firefly",
+	)
 
 	expectedSigningKeyRef := &core.VerifierRef{
 		Type:  core.VerifierTypeMSPIdentity,
@@ -2898,11 +2521,3 @@ func TestSubmitNetworkActionVersionError(t *testing.T) {
 	err := e.SubmitNetworkAction(context.Background(), "", signer, core.NetworkActionTerminate, location)
 	assert.Regexp(t, "FF10284", err)
 }
-
-func TestCallbacksWrongNamespace(t *testing.T) {
-	e, _ := newTestFabric()
-	nsOpID := "ns1:" + fftypes.NewUUID().String()
-	e.callbacks.OperationUpdate(context.Background(), e, nsOpID, core.OpStatusSucceeded, "tx123", "", nil)
-	e.callbacks.BatchPinComplete(context.Background(), &blockchain.BatchPin{Namespace: "ns1"}, nil)
-	e.callbacks.BlockchainNetworkAction(context.Background(), "ns1", "terminate", nil, nil, nil)
-}
diff --git a/internal/dataexchange/ffdx/ffdx_test.go b/internal/dataexchange/ffdx/ffdx_test.go
index d511bd51b..ac7a8189f 100644
--- a/internal/dataexchange/ffdx/ffdx_test.go
+++ b/internal/dataexchange/ffdx/ffdx_test.go
@@ -124,13 +124,13 @@ func TestInitMissingURL(t *testing.T) {
 
 func acker() func(args mock.Arguments) {
 	return func(args mock.Arguments) {
-		args[0].(dataexchange.DXEvent).Ack()
+		args[1].(dataexchange.DXEvent).Ack()
 	}
 }
 
 func manifestAcker(manifest string) func(args mock.Arguments) {
 	return func(args mock.Arguments) {
-		args[0].(dataexchange.DXEvent).AckWithManifest(manifest)
+		args[1].(dataexchange.DXEvent).AckWithManifest(manifest)
 	}
 }
 
@@ -505,7 +505,7 @@ func TestMessageEvents(t *testing.T) {
 	assert.NoError(t, err)
 
 	namespacedID1 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
+	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "1" &&
 			ev.NamespacedID() == namespacedID1 &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
@@ -518,7 +518,7 @@ func TestMessageEvents(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"1"}`, string(msg))
 
 	namespacedID2 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
+	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "2" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().TrackingID == namespacedID2 &&
@@ -529,7 +529,7 @@ func TestMessageEvents(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"2"}`, string(msg))
 
 	namespacedID3 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
+	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "3" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().TrackingID == namespacedID3 &&
@@ -541,7 +541,7 @@ func TestMessageEvents(t *testing.T) {
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"3"}`, string(msg))
 
-	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
+	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "4" &&
 			ev.Type() == dataexchange.DXEventTypeMessageReceived &&
 			ev.MessageReceived().PeerID == "peer1"
@@ -565,7 +565,7 @@ func TestBlobEvents(t *testing.T) {
 	assert.NoError(t, err)
 
 	namespacedID5 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
+	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "5" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().TrackingID == namespacedID5 &&
@@ -577,7 +577,7 @@ func TestBlobEvents(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"5"}`, string(msg))
 
 	namespacedID6 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
+	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "6" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().TrackingID == namespacedID6 &&
@@ -590,7 +590,7 @@ func TestBlobEvents(t *testing.T) {
 
 	u := fftypes.NewUUID()
 	hash := fftypes.NewRandB32()
-	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
+	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "9" &&
 			ev.Type() == dataexchange.DXEventTypePrivateBlobReceived &&
 			ev.PrivateBlobReceived().Hash.Equals(hash)
@@ -600,7 +600,7 @@ func TestBlobEvents(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"9"}`, string(msg))
 
 	namespacedID10 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
+	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "10" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().TrackingID == namespacedID10 &&
@@ -631,7 +631,7 @@ func TestEventsWithManifest(t *testing.T) {
 	h.SetHandler("ns1", mcb)
 
 	namespacedID1 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
+	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "1" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().Status == core.OpStatusPending
@@ -641,7 +641,7 @@ func TestEventsWithManifest(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"1"}`, string(msg))
 
 	namespacedID2 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
+	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "2" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().Status == core.OpStatusPending
diff --git a/internal/events/batch_pin_complete_test.go b/internal/events/batch_pin_complete_test.go
index 5213a3b48..fc7117c86 100644
--- a/internal/events/batch_pin_complete_test.go
+++ b/internal/events/batch_pin_complete_test.go
@@ -91,7 +91,6 @@ func TestBatchPinCompleteOkBroadcast(t *testing.T) {
 	data := &core.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
 	batch := sampleBatch(t, core.BatchTypeBroadcast, core.TransactionTypeBatchPin, core.DataArray{data})
 	batchPin := &blockchain.BatchPin{
-		Namespace:       "ns1",
 		TransactionID:   batch.Payload.TX.ID,
 		BatchID:         batch.ID,
 		BatchPayloadRef: "Qmf412jQZiuVUtdgnB36FXFX7xg5V6KEbSJ4dpQuhkLyfD",
@@ -137,7 +136,7 @@ func TestBatchPinCompleteOkBroadcast(t *testing.T) {
 	mdi.On("GetBatchByID", mock.Anything, "ns1", mock.Anything).Return(nil, nil)
 	msd.On("InitiateDownloadBatch", mock.Anything, batchPin.TransactionID, batchPin.BatchPayloadRef).Return(nil)
 
-	err := em.BatchPinComplete(batchPin, &core.VerifierRef{
+	err := em.BatchPinComplete("ns1", batchPin, &core.VerifierRef{
 		Type:  core.VerifierTypeEthAddress,
 		Value: "0x12345",
 	})
@@ -154,7 +153,6 @@ func TestBatchPinCompleteOkBroadcastExistingBatch(t *testing.T) {
 	data := &core.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
 	batch := sampleBatch(t, core.BatchTypeBroadcast, core.TransactionTypeBatchPin, core.DataArray{data})
 	batchPin := &blockchain.BatchPin{
-		Namespace:       "ns1",
 		TransactionID:   batch.Payload.TX.ID,
 		BatchID:         batch.ID,
 		BatchPayloadRef: "Qmf412jQZiuVUtdgnB36FXFX7xg5V6KEbSJ4dpQuhkLyfD",
@@ -206,7 +204,7 @@ func TestBatchPinCompleteOkBroadcastExistingBatch(t *testing.T) {
 	mdi.On("InsertPins", mock.Anything, mock.Anything).Return(nil).Once()
 	mdi.On("GetBatchByID", mock.Anything, "ns1", mock.Anything).Return(batchPersisted, nil)
 
-	err := em.BatchPinComplete(batchPin, &core.VerifierRef{
+	err := em.BatchPinComplete("ns1", batchPin, &core.VerifierRef{
 		Type:  core.VerifierTypeEthAddress,
 		Value: "0x12345",
 	})
@@ -221,7 +219,6 @@ func TestBatchPinCompleteOkPrivate(t *testing.T) {
 	defer cancel()
 
 	batchPin := &blockchain.BatchPin{
-		Namespace:     "ns1",
 		TransactionID: fftypes.NewUUID(),
 		BatchID:       fftypes.NewUUID(),
 		Contexts:      []*fftypes.Bytes32{fftypes.NewRandB32()},
@@ -243,7 +240,7 @@ func TestBatchPinCompleteOkPrivate(t *testing.T) {
 	mdi.On("InsertEvent", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("GetBatchByID", mock.Anything, "ns1", mock.Anything).Return(nil, nil)
 
-	err := em.BatchPinComplete(batchPin, &core.VerifierRef{
+	err := em.BatchPinComplete("ns1", batchPin, &core.VerifierRef{
 		Type:  core.VerifierTypeEthAddress,
 		Value: "0xffffeeee",
 	})
@@ -264,7 +261,6 @@ func TestBatchPinCompleteInsertPinsFail(t *testing.T) {
 	cancel()
 
 	batchPin := &blockchain.BatchPin{
-		Namespace:     "ns1",
 		TransactionID: fftypes.NewUUID(),
 		BatchID:       fftypes.NewUUID(),
 		Contexts:      []*fftypes.Bytes32{fftypes.NewRandB32()},
@@ -285,7 +281,7 @@ func TestBatchPinCompleteInsertPinsFail(t *testing.T) {
 	mth.On("InsertBlockchainEvent", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("InsertEvent", mock.Anything, mock.Anything).Return(nil)
 
-	err := em.BatchPinComplete(batchPin, &core.VerifierRef{
+	err := em.BatchPinComplete("ns1", batchPin, &core.VerifierRef{
 		Type:  core.VerifierTypeEthAddress,
 		Value: "0xffffeeee",
 	})
@@ -300,7 +296,6 @@ func TestBatchPinCompleteGetBatchByIDFails(t *testing.T) {
 	cancel()
 
 	batchPin := &blockchain.BatchPin{
-		Namespace:     "ns1",
 		TransactionID: fftypes.NewUUID(),
 		BatchID:       fftypes.NewUUID(),
 		Contexts:      []*fftypes.Bytes32{fftypes.NewRandB32()},
@@ -321,7 +316,7 @@ func TestBatchPinCompleteGetBatchByIDFails(t *testing.T) {
 	mdi.On("InsertEvent", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("GetBatchByID", mock.Anything, "ns1", mock.Anything).Return(nil, fmt.Errorf("batch lookup failed"))
 
-	err := em.BatchPinComplete(batchPin, &core.VerifierRef{
+	err := em.BatchPinComplete("ns1", batchPin, &core.VerifierRef{
 		Type:  core.VerifierTypeEthAddress,
 		Value: "0xffffeeee",
 	})
@@ -335,7 +330,6 @@ func TestSequencedBroadcastInitiateDownloadFail(t *testing.T) {
 	em, cancel := newTestEventManager(t)
 
 	batchPin := &blockchain.BatchPin{
-		Namespace:       "ns1",
 		TransactionID:   fftypes.NewUUID(),
 		BatchID:         fftypes.NewUUID(),
 		BatchPayloadRef: "Qmf412jQZiuVUtdgnB36FXFX7xg5V6KEbSJ4dpQuhkLyfD",
@@ -360,7 +354,7 @@ func TestSequencedBroadcastInitiateDownloadFail(t *testing.T) {
 	msd := em.sharedDownload.(*shareddownloadmocks.Manager)
 	msd.On("InitiateDownloadBatch", mock.Anything, batchPin.TransactionID, batchPin.BatchPayloadRef).Return(fmt.Errorf("pop"))
 
-	err := em.BatchPinComplete(batchPin, &core.VerifierRef{
+	err := em.BatchPinComplete("ns1", batchPin, &core.VerifierRef{
 		Type:  core.VerifierTypeEthAddress,
 		Value: "0xffffeeee",
 	})
@@ -376,7 +370,7 @@ func TestBatchPinCompleteNoTX(t *testing.T) {
 
 	batch := &blockchain.BatchPin{}
 
-	err := em.BatchPinComplete(batch, &core.VerifierRef{
+	err := em.BatchPinComplete("ns1", batch, &core.VerifierRef{
 		Type:  core.VerifierTypeEthAddress,
 		Value: "0x12345",
 	})
@@ -388,14 +382,13 @@ func TestBatchPinCompleteWrongNamespace(t *testing.T) {
 	defer cancel()
 
 	batch := &blockchain.BatchPin{
-		Namespace:     "ns2",
 		TransactionID: fftypes.NewUUID(),
 		Event: blockchain.Event{
 			BlockchainTXID: "0x12345",
 		},
 	}
 
-	err := em.BatchPinComplete(batch, &core.VerifierRef{
+	err := em.BatchPinComplete("ns2", batch, &core.VerifierRef{
 		Type:  core.VerifierTypeEthAddress,
 		Value: "0x12345",
 	})
@@ -408,14 +401,13 @@ func TestBatchPinCompleteNonMultiparty(t *testing.T) {
 	em.multiparty = nil
 
 	batch := &blockchain.BatchPin{
-		Namespace:     "ns1",
 		TransactionID: fftypes.NewUUID(),
 		Event: blockchain.Event{
 			BlockchainTXID: "0x12345",
 		},
 	}
 
-	err := em.BatchPinComplete(batch, &core.VerifierRef{
+	err := em.BatchPinComplete("ns1", batch, &core.VerifierRef{
 		Type:  core.VerifierTypeEthAddress,
 		Value: "0x12345",
 	})
diff --git a/internal/multiparty/operations_test.go b/internal/multiparty/operations_test.go
index c37772b27..9c1fb12e8 100644
--- a/internal/multiparty/operations_test.go
+++ b/internal/multiparty/operations_test.go
@@ -42,6 +42,7 @@ func TestPrepareAndRunBatchPin(t *testing.T) {
 			SignerRef: core.SignerRef{
 				Key: "0x123",
 			},
+			Namespace: "ns1",
 		},
 	}
 	contexts := []*fftypes.Bytes32{
@@ -51,7 +52,7 @@ func TestPrepareAndRunBatchPin(t *testing.T) {
 	addBatchPinInputs(op, batch.ID, contexts, "payload1")
 
 	mp.mdi.On("GetBatchByID", context.Background(), "ns1", batch.ID).Return(batch, nil)
-	mp.mbi.On("SubmitBatchPin", context.Background(), "ns1:"+op.ID.String(), "0x123", mock.Anything, mock.Anything).Return(nil)
+	mp.mbi.On("SubmitBatchPin", context.Background(), "ns1:"+op.ID.String(), "ns1", "0x123", mock.Anything, mock.Anything).Return(nil)
 
 	po, err := mp.PrepareOperation(context.Background(), op)
 	assert.NoError(t, err)
@@ -191,6 +192,7 @@ func TestRunBatchPinV1(t *testing.T) {
 			SignerRef: core.SignerRef{
 				Key: "0x123",
 			},
+			Namespace: "ns1",
 		},
 	}
 	contexts := []*fftypes.Bytes32{
@@ -199,7 +201,7 @@ func TestRunBatchPinV1(t *testing.T) {
 	}
 	addBatchPinInputs(op, batch.ID, contexts, "payload1")
 
-	mp.mbi.On("SubmitBatchPin", context.Background(), "ns1:"+op.ID.String(), "0x123", mock.Anything, mock.Anything).Return(nil)
+	mp.mbi.On("SubmitBatchPin", context.Background(), "ns1:"+op.ID.String(), "ns1", "0x123", mock.Anything, mock.Anything).Return(nil)
 
 	_, complete, err := mp.RunOperation(context.Background(), opBatchPin(op, batch, contexts, "payload1"))
 
diff --git a/internal/operations/manager_test.go b/internal/operations/manager_test.go
index c53bd8db9..50072a95f 100644
--- a/internal/operations/manager_test.go
+++ b/internal/operations/manager_test.go
@@ -18,7 +18,6 @@ package operations
 import (
 	"context"
 	"fmt"
-	"strings"
 	"testing"
 
 	"github.com/hyperledger/firefly-common/pkg/config"
@@ -26,11 +25,9 @@ import (
 	"github.com/hyperledger/firefly/internal/coreconfig"
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/mocks/databasemocks"
-	"github.com/hyperledger/firefly/mocks/dataexchangemocks"
 	"github.com/hyperledger/firefly/mocks/datamocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
-	"github.com/hyperledger/firefly/pkg/dataexchange"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 )
@@ -410,141 +407,6 @@ func TestWriteOperationFailure(t *testing.T) {
 	mdi.AssertExpectations(t)
 }
 
-func TestTransferResultManifestMismatch(t *testing.T) {
-	om, cancel := newTestOperations(t)
-	defer cancel()
-	om.updater.conf.workerCount = 0
-
-	opID1 := fftypes.NewUUID()
-	mdi := om.database.(*databasemocks.Plugin)
-	mdi.On("GetOperations", mock.Anything, "ns1", mock.Anything).Return([]*core.Operation{
-		{
-			ID:        opID1,
-			Namespace: "ns1",
-			Type:      core.OpTypeDataExchangeSendBatch,
-			Input: fftypes.JSONObject{
-				"batch": fftypes.NewUUID().String(),
-			},
-		},
-	}, nil, nil)
-	mdi.On("ResolveOperation", mock.Anything, "ns1", opID1, core.OpStatusFailed, mock.MatchedBy(func(errorMsg *string) bool {
-		return strings.Contains(*errorMsg, "FF10329")
-	}), fftypes.JSONObject{
-		"extra": "info",
-	}).Return(nil)
-	mdi.On("GetBatchByID", mock.Anything, "ns1", mock.Anything).Return(&core.BatchPersisted{
-		Manifest: fftypes.JSONAnyPtr("my-manifest"),
-	}, nil)
-
-	mdx := &dataexchangemocks.Plugin{}
-	mdx.On("Name").Return("utdx")
-	mdx.On("Capabilities").Return(&dataexchange.Capabilities{
-		Manifest: true,
-	})
-	mde := &dataexchangemocks.DXEvent{}
-	mde.On("NamespacedID").Return("ns1:" + opID1.String())
-	mde.On("Ack").Return()
-	mde.On("TransferResult").Return(&dataexchange.TransferResult{
-		TrackingID: opID1.String(),
-		Status:     core.OpStatusSucceeded,
-		TransportStatusUpdate: core.TransportStatusUpdate{
-			Info:     fftypes.JSONObject{"extra": "info"},
-			Manifest: "Sally",
-		},
-	})
-	om.TransferResult(mdx, mde)
-
-	mde.AssertExpectations(t)
-	mdi.AssertExpectations(t)
-
-}
-
-func TestTransferResultHashMismatch(t *testing.T) {
-
-	om, cancel := newTestOperations(t)
-	cancel()
-	om.updater.conf.workerCount = 0
-
-	opID1 := fftypes.NewUUID()
-	mdi := om.database.(*databasemocks.Plugin)
-	mdi.On("GetOperations", mock.Anything, "ns1", mock.Anything).Return([]*core.Operation{
-		{
-			ID:        opID1,
-			Namespace: "ns1",
-			Type:      core.OpTypeDataExchangeSendBlob,
-			Input: fftypes.JSONObject{
-				"hash": "Bob",
-			},
-		},
-	}, nil, nil)
-	mdi.On("ResolveOperation", mock.Anything, "ns1", opID1, core.OpStatusFailed, mock.MatchedBy(func(errorMsg *string) bool {
-		return strings.Contains(*errorMsg, "FF10348")
-	}), fftypes.JSONObject{
-		"extra": "info",
-	}).Return(nil)
-
-	mdx := &dataexchangemocks.Plugin{}
-	mdx.On("Name").Return("utdx")
-	mdx.On("Capabilities").Return(&dataexchange.Capabilities{
-		Manifest: true,
-	})
-	mde := &dataexchangemocks.DXEvent{}
-	mde.On("NamespacedID").Return("ns1:" + opID1.String())
-	mde.On("Ack").Return()
-	mde.On("TransferResult").Return(&dataexchange.TransferResult{
-		TrackingID: opID1.String(),
-		Status:     core.OpStatusSucceeded,
-		TransportStatusUpdate: core.TransportStatusUpdate{
-			Info: fftypes.JSONObject{"extra": "info"},
-			Hash: "Sally",
-		},
-	})
-	om.TransferResult(mdx, mde)
-
-	mde.AssertExpectations(t)
-	mdi.AssertExpectations(t)
-
-}
-
-func TestTransferResultBatchLookupFail(t *testing.T) {
-	om, cancel := newTestOperations(t)
-	cancel()
-	om.updater.conf.workerCount = 0
-
-	opID1 := fftypes.NewUUID()
-	mdi := om.database.(*databasemocks.Plugin)
-	mdi.On("GetOperations", mock.Anything, "ns1", mock.Anything).Return([]*core.Operation{
-		{
-			ID:   opID1,
-			Type: core.OpTypeDataExchangeSendBatch,
-			Input: fftypes.JSONObject{
-				"batch": fftypes.NewUUID().String(),
-			},
-		},
-	}, nil, nil)
-	mdi.On("GetBatchByID", mock.Anything, "ns1", mock.Anything).Return(nil, fmt.Errorf("pop"))
-
-	mdx := &dataexchangemocks.Plugin{}
-	mdx.On("Name").Return("utdx")
-	mdx.On("Capabilities").Return(&dataexchange.Capabilities{
-		Manifest: true,
-	})
-	mde := &dataexchangemocks.DXEvent{}
-	mde.On("NamespacedID").Return("ns1:" + opID1.String())
-	mde.On("TransferResult").Return(&dataexchange.TransferResult{
-		TrackingID: opID1.String(),
-		Status:     core.OpStatusSucceeded,
-		TransportStatusUpdate: core.TransportStatusUpdate{
-			Info:     fftypes.JSONObject{"extra": "info"},
-			Manifest: "Sally",
-		},
-	})
-	om.TransferResult(mdx, mde)
-
-	mdi.AssertExpectations(t)
-
-}
-
 func TestResolveOperationByNamespacedIDOk(t *testing.T) {
 	om, cancel := newTestOperations(t)
 	defer cancel()
diff --git a/internal/orchestrator/bound_callbacks_test.go b/internal/orchestrator/bound_callbacks_test.go
index 2697f2ad1..6079b6ea9 100644
--- a/internal/orchestrator/bound_callbacks_test.go
+++ b/internal/orchestrator/bound_callbacks_test.go
@@ -17,6 +17,7 @@
 package orchestrator
 
 import (
+	"context"
 	"fmt"
 	"testing"
 
@@ -44,32 +45,107 @@ func TestBoundCallbacks(t *testing.T) {
 	info := fftypes.JSONObject{"hello": "world"}
 	hash := fftypes.NewRandB32()
 	opID := fftypes.NewUUID()
-
 	nsOpID := "ns1:" + opID.String()
+
 	mom.On("SubmitOperationUpdate", mock.Anything, &operations.OperationUpdate{
 		NamespacedOpID: nsOpID,
 		Status:         core.OpStatusFailed,
 		BlockchainTXID: "0xffffeeee",
 		ErrorMessage:   "error info",
 		Output:         info,
-	}).Return()
-
+	}).Return().Once()
 	bc.OperationUpdate(mbi, nsOpID, core.OpStatusFailed, "0xffffeeee", "error info", info)
 
-	mde := &dataexchangemocks.DXEvent{}
-	mom.On("TransferResult", mdx, mde).Return()
-	mei.On("DXEvent", mdx, mde).Return()
-
-	mde.On("Type").Return(dataexchange.DXEventTypeTransferResult).Once()
-	bc.DXEvent(mde)
-
-	mde.On("Type").Return(dataexchange.DXEventTypeMessageReceived).Once()
-	bc.DXEvent(mde)
-
 	mei.On("SharedStorageBatchDownloaded", mss, "payload1", []byte(`{}`)).Return(nil, fmt.Errorf("pop"))
 	_, err := bc.SharedStorageBatchDownloaded("payload1", []byte(`{}`))
 	assert.EqualError(t, err, "pop")
 
 	mei.On("SharedStorageBlobDownloaded", mss, *hash, int64(12345), "payload1").Return()
 	bc.SharedStorageBlobDownloaded(*hash, 12345, "payload1")
+
+	mei.AssertExpectations(t)
+	mbi.AssertExpectations(t)
+	mdx.AssertExpectations(t)
+	mss.AssertExpectations(t)
+	mom.AssertExpectations(t)
+}
+
+func TestBoundCallbacksDXEvent(t *testing.T) {
+	mei := &eventmocks.EventManager{}
+	mdx := &dataexchangemocks.Plugin{}
+	mss := &sharedstoragemocks.Plugin{}
+	mom := &operationmocks.Manager{}
+	bc := boundCallbacks{dx: mdx, ei: mei, ss: mss, om: mom}
+
+	ctx := context.Background()
+	info := fftypes.JSONObject{"hello": "world"}
+	opID := fftypes.NewUUID()
+	nsOpID := "ns1:" + opID.String()
+
+	mdx.On("Capabilities").Return(&dataexchange.Capabilities{
+		Manifest: true,
+	})
+
+	event1 := &dataexchangemocks.DXEvent{}
+	mei.On("DXEvent", mdx, event1).Return().Once()
+	event1.On("Type").Return(dataexchange.DXEventTypeMessageReceived).Once()
+	bc.DXEvent(ctx, event1)
+	event1.AssertExpectations(t)
+
+	event2 := &dataexchangemocks.DXEvent{}
+	event2.On("Type").Return(dataexchange.DXEventTypeTransferResult).Once()
+	event2.On("TransferResult").Return(&dataexchange.TransferResult{
+		TrackingID: opID.String(),
+		Status:     core.OpStatusSucceeded,
+		TransportStatusUpdate: core.TransportStatusUpdate{
+			Info:     info,
+			Manifest: "Sally",
+		},
+	})
+	event2.On("NamespacedID").Return(nsOpID)
+	event2.On("Ack").Return()
+	mom.On("SubmitOperationUpdate", mock.Anything, mock.MatchedBy(func(update *operations.OperationUpdate) bool {
+		if update.NamespacedOpID == nsOpID &&
+			update.Status == core.OpStatusSucceeded &&
+			update.VerifyManifest &&
+			update.DXManifest == "Sally" &&
+			update.DXHash == "" {
+			update.OnComplete()
+			return true
+		}
+		return false
+	})).Return().Once()
+	bc.DXEvent(ctx, event2)
+	event2.AssertExpectations(t)
+
+	event3 := &dataexchangemocks.DXEvent{}
+	event3.On("Type").Return(dataexchange.DXEventTypeTransferResult).Once()
+	event3.On("TransferResult").Return(&dataexchange.TransferResult{
+		TrackingID: opID.String(),
+		Status:     core.OpStatusSucceeded,
+		TransportStatusUpdate: core.TransportStatusUpdate{
+			Info: info,
+			Hash: "hash1",
+		},
+	})
+	event3.On("NamespacedID").Return(nsOpID)
+	event3.On("Ack").Return()
+	mom.On("SubmitOperationUpdate", mock.Anything, mock.MatchedBy(func(update *operations.OperationUpdate) bool {
+		if update.NamespacedOpID == nsOpID &&
+			update.Status == core.OpStatusSucceeded &&
+			update.VerifyManifest &&
+			update.DXManifest == "" &&
+			update.DXHash == "hash1" {
+			update.OnComplete()
+			return true
+		}
+		return false
+	})).Return().Once()
+	bc.DXEvent(ctx, event3)
+	event3.AssertExpectations(t)
+
+	mei.AssertExpectations(t)
+	mdx.AssertExpectations(t)
+	mss.AssertExpectations(t)
+	mom.AssertExpectations(t)
 }
diff --git a/internal/orchestrator/orchestrator_test.go b/internal/orchestrator/orchestrator_test.go
index 33c697b47..dfd7c0cde 100644
--- a/internal/orchestrator/orchestrator_test.go
+++ b/internal/orchestrator/orchestrator_test.go
@@ -205,12 +205,12 @@ func TestInitOK(t *testing.T) {
 	defer or.cleanup(t)
 	or.namespace.RemoteName = "ns2"
 	or.mdi.On("SetHandler", "ns", mock.Anything).Return()
-	or.mbi.On("SetHandler", "ns2", mock.Anything).Return()
+	or.mbi.On("SetHandler", "ns", mock.Anything).Return()
 	or.mbi.On("SetOperationHandler", "ns", mock.Anything).Return()
 	or.mdi.On("GetIdentities", mock.Anything, "ns", mock.Anything).Return([]*core.Identity{{}}, nil, nil)
 	or.mdx.On("SetHandler", "ns2", mock.Anything).Return()
 	or.mdx.On("SetNodes", mock.Anything).Return()
-	or.mps.On("SetHandler", "ns2", mock.Anything).Return()
+	or.mps.On("SetHandler", "ns", mock.Anything).Return()
 	or.mti.On("SetHandler", "ns", mock.Anything).Return(nil)
 	or.mti.On("SetOperationHandler", "ns", mock.Anything).Return(nil)
 	or.mmp.On("ConfigureContract", mock.Anything, mock.Anything).Return(nil)
diff --git a/test/e2e/client/restclient.go b/test/e2e/client/restclient.go
index bdc938fe5..6d34bc7c8 100644
--- a/test/e2e/client/restclient.go
+++ b/test/e2e/client/restclient.go
@@ -72,6 +72,7 @@ var (
 	urlNodesSelf         = "/network/nodes/self"
 	urlNetworkAction     = "/network/action"
 	urlGetOrgKeys        = "/identities/%s/verifiers"
+	urlStatus            = "/status"
 )
 
 type Logger interface {
@@ -924,3 +925,12 @@ func (client *FireFlyClient) NetworkAction(t *testing.T, action core.NetworkActi
 	require.NoError(t, err)
 	require.Equal(t, 202, resp.StatusCode(), "POST %s [%d]: %s", path, resp.StatusCode(), resp.String())
 }
+
+func (client *FireFlyClient) GetStatus() (*core.NamespaceStatus, *resty.Response, error) {
+	var status core.NamespaceStatus
+	path := client.namespaced(urlStatus)
+	resp, err := client.Client.R().
+		SetResult(&status).
+		Get(path)
+	return &status, resp, err
+}
diff --git a/test/e2e/e2e.go b/test/e2e/e2e.go
index 29cdcd969..93cb06d44 100644
--- a/test/e2e/e2e.go
+++ b/test/e2e/e2e.go
@@ -62,7 +62,7 @@ func PollForUp(t *testing.T, client *client.FireFlyClient) {
 	var resp *resty.Response
 	var err error
 	for i := 0; i < 3; i++ {
-		resp, err = client.GetNamespaces()
+		_, resp, err = client.GetStatus()
 		if err == nil && resp.StatusCode() == 200 {
 			break
 		}
diff --git a/test/e2e/multiparty/namespace_alias.go b/test/e2e/multiparty/namespace_alias.go
new file mode 100644
index 000000000..7f6089cd1
--- /dev/null
+++ b/test/e2e/multiparty/namespace_alias.go
@@ -0,0 +1,146 @@
+// Copyright © 2022 Kaleido, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package multiparty
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/hyperledger/firefly-common/pkg/fftypes"
+	"github.com/hyperledger/firefly/pkg/core"
+	"github.com/hyperledger/firefly/test/e2e"
+	"github.com/hyperledger/firefly/test/e2e/client"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+)
+
+type NamespaceAliasSuite struct {
+	suite.Suite
+	testState   *testState
+	stackName   string
+	adminHost1  string
+	adminHost2  string
+	configFile1 string
+	configFile2 string
+}
+
+func (suite *NamespaceAliasSuite) SetupSuite() {
+	suite.testState = beforeE2ETest(suite.T())
+	stack := e2e.ReadStack(suite.T())
+	suite.stackName = stack.Name
+
+	adminProtocol1 := schemeHTTP
+	if stack.Members[0].UseHTTPS {
+		adminProtocol1 = schemeHTTPS
+	}
+	adminProtocol2 := schemeHTTP
+	if stack.Members[1].UseHTTPS {
+		adminProtocol2 = schemeHTTPS
+	}
+	suite.adminHost1 = fmt.Sprintf("%s://%s:%d", adminProtocol1, stack.Members[0].FireflyHostname, stack.Members[0].ExposedAdminPort)
+	suite.adminHost2 = fmt.Sprintf("%s://%s:%d", adminProtocol2, stack.Members[1].FireflyHostname, stack.Members[1].ExposedAdminPort)
+
+	stackDir := os.Getenv("STACK_DIR")
+	if stackDir == "" {
+		suite.T().Fatal("STACK_DIR must be set")
+	}
+	suite.configFile1 = filepath.Join(stackDir, "runtime", "config", "firefly_core_0.yml")
+	suite.configFile2 = filepath.Join(stackDir, "runtime", "config", "firefly_core_1.yml")
+}
+
+func (suite *NamespaceAliasSuite) BeforeTest(suiteName, testName string) {
+	suite.testState = beforeE2ETest(suite.T())
+}
+
+func (suite *NamespaceAliasSuite) AfterTest(suiteName, testName string) {
+	e2e.VerifyAllOperationsSucceeded(suite.T(), []*client.FireFlyClient{suite.testState.client1, suite.testState.client2}, suite.testState.startTime)
+}
+
+func (suite *NamespaceAliasSuite) TestNamespaceMapping() {
+	defer suite.testState.Done()
+
+	address := deployContract(suite.T(), suite.stackName, "firefly/Firefly.json")
+	localNamespace1 := randomName(suite.T())
+	localNamespace2 := randomName(suite.T())
+	remoteNamespace := randomName(suite.T())
+	suite.T().Logf("Test namespace: local1=%s, local2=%s, remote=%s", localNamespace1, localNamespace2, remoteNamespace)
+
+	org := map[string]interface{}{}
+	namespaceInfo := map[string]interface{}{
+		"remotename": remoteNamespace,
+		"multiparty": map[string]interface{}{
+			"enabled": true,
+			"org":     org,
+			"contract": []map[string]interface{}{
+				{
+					"location": map[string]interface{}{"address": address},
+				},
+			},
+		},
+	}
+	data := &core.DataRefOrValue{Value: fftypes.JSONAnyPtr(`"test"`)}
+
+	// Add the new namespace to both config files
+	data1 := readConfig(suite.T(), suite.configFile1)
+	namespaceInfo["name"] = localNamespace1
+	org["name"] = suite.testState.org1.Name
+	org["key"] = suite.testState.org1key.Value
+	addNamespace(data1, namespaceInfo)
+	writeConfig(suite.T(), suite.configFile1, data1)
+
+	data2 := readConfig(suite.T(), suite.configFile2)
+	namespaceInfo["name"] = localNamespace2
+	org["name"] = suite.testState.org2.Name
+	org["key"] = suite.testState.org2key.Value
+	addNamespace(data2, namespaceInfo)
+	writeConfig(suite.T(), suite.configFile2, data2)
+
+	admin1 := client.NewResty(suite.T())
+	admin2 := client.NewResty(suite.T())
+	admin1.SetBaseURL(suite.adminHost1 + "/spi/v1")
+	admin2.SetBaseURL(suite.adminHost2 + "/spi/v1")
+
+	// Reset both nodes to pick up the new namespace
+	resetFireFly(suite.T(), admin1)
+	resetFireFly(suite.T(), admin2)
+	e2e.PollForUp(suite.T(), suite.testState.client1)
+	e2e.PollForUp(suite.T(), suite.testState.client2)
+
+	client1 := client.NewFireFly(suite.T(), suite.testState.client1.Hostname, localNamespace1)
+	client2 := client.NewFireFly(suite.T(), suite.testState.client2.Hostname, localNamespace2)
+
+	eventNames := "message_confirmed|blockchain_event_received"
+	queryString := fmt.Sprintf("namespace=%s&ephemeral&autoack&filter.events=%s", localNamespace1, eventNames)
+	received1 := e2e.WsReader(client1.WebSocket(suite.T(), queryString, nil))
+	queryString = fmt.Sprintf("namespace=%s&ephemeral&autoack&filter.events=%s", localNamespace2, eventNames)
+	received2 := e2e.WsReader(client2.WebSocket(suite.T(), queryString, nil))
+
+	// Register org/node identities on the new namespace
+	client1.RegisterSelfOrg(suite.T(), true)
+	client1.RegisterSelfNode(suite.T(), true)
+	client2.RegisterSelfOrg(suite.T(), true)
+	client2.RegisterSelfNode(suite.T(), true)
+
+	// Verify that a broadcast on the new namespace succeeds
+	resp, err := client1.BroadcastMessage(suite.T(), "topic", data, false)
+	require.NoError(suite.T(), err)
+	assert.Equal(suite.T(), 202, resp.StatusCode())
+	e2e.WaitForMessageConfirmed(suite.T(), received1, core.MessageTypeBroadcast)
+	e2e.WaitForMessageConfirmed(suite.T(), received2, core.MessageTypeBroadcast)
+}
diff --git a/test/e2e/runners/ethereum_multiparty_test.go b/test/e2e/runners/ethereum_multiparty_test.go
index 611661234..2c1221fb8 100644
--- a/test/e2e/runners/ethereum_multiparty_test.go
+++ b/test/e2e/runners/ethereum_multiparty_test.go
@@ -29,4 +29,5 @@ func TestEthereumMultipartyE2ESuite(t *testing.T) {
 	suite.Run(t, new(multiparty.TokensTestSuite))
 	suite.Run(t, new(multiparty.EthereumContractTestSuite))
 	suite.Run(t, new(multiparty.ContractMigrationTestSuite))
+	suite.Run(t, new(multiparty.NamespaceAliasSuite))
 }