Repository: hyperledger/firefly (go)

Issue:
Create operation and transaction for custom smart contract calls
Invoking a smart contract should be tracked by an `Operation` and a `Transaction` in FireFly.

Pull request description:
Create a Transaction and Operation for contract invoke requests
Resolves #393

Patch:
diff --git a/docs/swagger/swagger.yaml b/docs/swagger/swagger.yaml
index 989140d56..cb1984671 100644
--- a/docs/swagger/swagger.yaml
+++ b/docs/swagger/swagger.yaml
@@ -5252,6 +5252,7 @@ paths:
                     - token_create_pool
                     - token_announce_pool
                     - token_transfer
+                    - contract_invoke
                     type: string
                   updated: {}
                 type: object
@@ -5301,6 +5302,7 @@ paths:
                     - batch_pin
                     - token_pool
                     - token_transfer
+                    - contract_invoke
                     type: string
                 type: object
           description: Success
@@ -6010,6 +6012,7 @@ paths:
                     - token_create_pool
                     - token_announce_pool
                     - token_transfer
+                    - contract_invoke
                     type: string
                   updated: {}
                 type: object
@@ -6075,6 +6078,7 @@ paths:
                     - token_create_pool
                     - token_announce_pool
                     - token_transfer
+                    - contract_invoke
                     type: string
                   updated: {}
                 type: object
@@ -9976,6 +9980,7 @@ paths:
                     - batch_pin
                     - token_pool
                     - token_transfer
+                    - contract_invoke
                     type: string
                 type: object
           description: Success
@@ -10083,6 +10088,7 @@ paths:
                     - batch_pin
                     - token_pool
                     - token_transfer
+                    - contract_invoke
                     type: string
                 type: object
           description: Success
@@ -10132,6 +10138,7 @@ paths:
                       - batch_pin
                       - token_pool
                       - token_transfer
+                      - contract_invoke
                       type: string
                   type: object
                 type: array
diff --git a/internal/contracts/manager.go b/internal/contracts/manager.go
index 82d1c22af..120b4f62b 100644
--- a/internal/contracts/manager.go
+++ b/internal/contracts/manager.go
@@ -149,10 +149,37 @@ func (cm *contractManager) InvokeContract(ctx context.Context, ns string, req *f
 		return nil, err
 	}
 
-	if req.Method, err = cm.resolveInvokeContractRequest(ctx, ns, req); err != nil {
-		return nil, err
-	}
-	if err := cm.validateInvokeContractRequest(ctx, req); err != nil {
+	var op *fftypes.Operation
+	err = cm.database.RunAsGroup(ctx, func(ctx context.Context) (err error) {
+		if req.Method, err = cm.resolveInvokeContractRequest(ctx, ns, req); err != nil {
+			return err
+		}
+		if err := cm.validateInvokeContractRequest(ctx, req); err != nil {
+			return err
+		}
+
+		tx := &fftypes.Transaction{
+			ID:        fftypes.NewUUID(),
+			Namespace: ns,
+			Type:      fftypes.TransactionTypeContractInvoke,
+			Created:   fftypes.Now(),
+			Status:    fftypes.OpStatusPending,
+		}
+		if err := cm.database.UpsertTransaction(ctx, tx); err != nil {
+			return err
+		}
+
+		op = fftypes.NewTXOperation(
+			cm.blockchain,
+			ns,
+			tx.ID,
+			"",
+			fftypes.OpTypeContractInvoke,
+			fftypes.OpStatusPending)
+		op.Input = req.Input
+		return cm.database.InsertOperation(ctx, op)
+	})
+	if err != nil {
 		return nil, err
 	}
 
diff --git a/internal/database/sqlcommon/message_sql.go b/internal/database/sqlcommon/message_sql.go
index dbb70c8ec..d2df09270 100644
--- a/internal/database/sqlcommon/message_sql.go
+++ b/internal/database/sqlcommon/message_sql.go
@@ -183,34 +183,6 @@ func (s *SQLCommon) UpsertMessage(ctx context.Context, message *fftypes.Message,
 	return s.commitTx(ctx, tx, autoCommit)
 }
 
-// In SQL update+bump is a delete+insert within a TX
-func (s *SQLCommon) ReplaceMessage(ctx context.Context, message *fftypes.Message) (err error) {
-	ctx, tx, autoCommit, err := s.beginOrUseTx(ctx)
-	if err != nil {
-		return err
-	}
-	defer s.rollbackTx(ctx, tx, autoCommit)
-
-	if err := s.deleteTx(ctx, tx,
-		sq.Delete("messages").
-			Where(sq.And{
-				sq.Eq{"id": message.Header.ID},
-			}),
-		nil, // no change event
-	); err != nil {
-		return err
-	}
-
-	if err = s.attemptMessageInsert(ctx, tx, message); err != nil {
-		return err
-	}
-
-	// Note there is no call to updateMessageDataRefs as the data refs are not allowed to change,
-	// and are correlated by UUID (not sequence)
-
-	return s.commitTx(ctx, tx, autoCommit)
-}
-
 func (s *SQLCommon) updateMessageDataRefs(ctx context.Context, tx *txWrapper, message *fftypes.Message, recreateDatarefs bool) error {
 
 	if recreateDatarefs {
diff --git a/internal/events/operation_update.go b/internal/events/operation_update.go
index c6bac4129..cdb4cc29c 100644
--- a/internal/events/operation_update.go
+++ b/internal/events/operation_update.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2021 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -39,11 +39,6 @@ func (em *eventManager) OperationUpdate(plugin fftypes.Named, operationID *fftyp
 
 	// Special handling for OpTypeTokenTransfer, which writes an event when it fails
 	if op.Type == fftypes.OpTypeTokenTransfer && txState == fftypes.OpStatusFailed {
-		txUpdate := database.TransactionQueryFactory.NewUpdate(em.ctx).Set("status", txState)
-		if err := em.database.UpdateTransaction(em.ctx, op.Transaction, txUpdate); err != nil {
-			return err
-		}
-
 		event := fftypes.NewEvent(fftypes.EventTypeTransferOpFailed, op.Namespace, op.ID)
 		if err := em.database.InsertEvent(em.ctx, event); err != nil {
 			return err
diff --git a/internal/events/tokens_transferred.go b/internal/events/tokens_transferred.go
index fc126c20c..1082629ac 100644
--- a/internal/events/tokens_transferred.go
+++ b/internal/events/tokens_transferred.go
@@ -133,7 +133,7 @@ func (em *eventManager) TokensTransferred(ti tokens.Plugin, transfer *tokens.Tok
 					if msg.State == fftypes.MessageStateStaged {
 						// Message can now be sent
 						msg.State = fftypes.MessageStateReady
-						if err := em.database.ReplaceMessage(ctx, msg); err != nil {
+						if err := em.database.UpsertMessage(ctx, msg, database.UpsertOptimizationExisting); err != nil {
 							return err
 						}
 					} else {
diff --git a/mocks/databasemocks/plugin.go b/mocks/databasemocks/plugin.go
index 65d41e5a4..92a7fb359 100644
--- a/mocks/databasemocks/plugin.go
+++ b/mocks/databasemocks/plugin.go
@@ -2199,20 +2199,6 @@ func (_m *Plugin) Name() string {
 	return r0
 }
 
-// ReplaceMessage provides a mock function with given fields: ctx, message
-func (_m *Plugin) ReplaceMessage(ctx context.Context, message *fftypes.Message) error {
-	ret := _m.Called(ctx, message)
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.Message) error); ok {
-		r0 = rf(ctx, message)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
 // RunAsGroup provides a mock function with given fields: ctx, fn
 func (_m *Plugin) RunAsGroup(ctx context.Context, fn func(context.Context) error) error {
 	ret := _m.Called(ctx, fn)
diff --git a/pkg/database/plugin.go b/pkg/database/plugin.go
index 33698aeb1..28fd215a0 100644
--- a/pkg/database/plugin.go
+++ b/pkg/database/plugin.go
@@ -80,10 +80,6 @@ type iMessageCollection interface {
 	// UpdateMessage - Update message
 	UpdateMessage(ctx context.Context, id *fftypes.UUID, update Update) (err error)
 
-	// ReplaceMessage updates the message, and assigns it a new sequence number at the front of the list.
-	// A new event is raised for the message, with the new sequence number - as if it was brand new.
-	ReplaceMessage(ctx context.Context, message *fftypes.Message) (err error)
-
 	// UpdateMessages - Update messages
 	UpdateMessages(ctx context.Context, filter Filter, update Update) (err error)
 
diff --git a/pkg/fftypes/operation.go b/pkg/fftypes/operation.go
index e8537c58a..472fdc75b 100644
--- a/pkg/fftypes/operation.go
+++ b/pkg/fftypes/operation.go
@@ -1,4 +1,4 @@
-// Copyright © 2021 Kaleido, Inc.
+// Copyright © 2022 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -36,6 +36,8 @@ var (
 	OpTypeTokenAnnouncePool OpType = ffEnum("optype", "token_announce_pool")
 	// OpTypeTokenTransfer is a token transfer
 	OpTypeTokenTransfer OpType = ffEnum("optype", "token_transfer")
+	// OpTypeContractInvoke is a smart contract invoke
+	OpTypeContractInvoke OpType = ffEnum("optype", "contract_invoke")
 )
 
 // OpStatus is the current status of an operation
diff --git a/pkg/fftypes/transaction.go b/pkg/fftypes/transaction.go
index 7032db192..a9d539e9e 100644
--- a/pkg/fftypes/transaction.go
+++ b/pkg/fftypes/transaction.go
@@ -27,6 +27,8 @@ var (
 	TransactionTypeTokenPool TransactionType = ffEnum("txtype", "token_pool")
 	// TransactionTypeTokenTransfer represents a token transfer
 	TransactionTypeTokenTransfer TransactionType = ffEnum("txtype", "token_transfer")
+	// TransactionTypeContractInvoke is a smart contract invoke
+	TransactionTypeContractInvoke OpType = ffEnum("txtype", "contract_invoke")
 )
 
 // TransactionRef refers to a transaction, in other types


Test patch:
diff --git a/internal/contracts/manager_test.go b/internal/contracts/manager_test.go
index 683fb834a..7cd2fc377 100644
--- a/internal/contracts/manager_test.go
+++ b/internal/contracts/manager_test.go
@@ -40,6 +40,8 @@ func newTestContractManager() *contractManager {
 	mim := &identitymanagermocks.Manager{}
 	mbi := &blockchainmocks.Plugin{}
 
+	mbi.On("Name").Return("mockblockchain").Maybe()
+
 	rag := mdb.On("RunAsGroup", mock.Anything, mock.Anything).Maybe()
 	rag.RunFn = func(a mock.Arguments) {
 		rag.ReturnArguments = mock.Arguments{
@@ -1070,6 +1072,7 @@ func TestInvokeContract(t *testing.T) {
 	cm := newTestContractManager()
 	mbi := cm.blockchain.(*blockchainmocks.Plugin)
 	mim := cm.identity.(*identitymanagermocks.Manager)
+	mdi := cm.database.(*databasemocks.Plugin)
 
 	req := &fftypes.ContractCallRequest{
 		Type:      fftypes.CallTypeInvoke,
@@ -1085,6 +1088,12 @@ func TestInvokeContract(t *testing.T) {
 	}
 
 	mim.On("ResolveSigningKey", mock.Anything, "").Return("key-resolved", nil)
+	mdi.On("UpsertTransaction", mock.Anything, mock.MatchedBy(func(tx *fftypes.Transaction) bool {
+		return tx.Namespace == "ns1" && tx.Type == fftypes.TransactionTypeContractInvoke
+	})).Return(nil)
+	mdi.On("InsertOperation", mock.Anything, mock.MatchedBy(func(op *fftypes.Operation) bool {
+		return op.Namespace == "ns1" && op.Type == fftypes.OpTypeContractInvoke && op.Plugin == "mockblockchain"
+	})).Return(nil)
 	mbi.On("InvokeContract", mock.Anything, mock.AnythingOfType("*fftypes.UUID"), "key-resolved", req.Location, req.Method, req.Input).Return(struct{}{}, nil)
 
 	_, err := cm.InvokeContract(context.Background(), "ns1", req)
@@ -1130,9 +1139,36 @@ func TestInvokeContractFailResolve(t *testing.T) {
 	assert.Regexp(t, "FF10313", err)
 }
 
+func TestInvokeContractTXFail(t *testing.T) {
+	cm := newTestContractManager()
+	mim := cm.identity.(*identitymanagermocks.Manager)
+	mdi := cm.database.(*databasemocks.Plugin)
+
+	req := &fftypes.ContractCallRequest{
+		Type:      fftypes.CallTypeInvoke,
+		Interface: fftypes.NewUUID(),
+		Ledger:    fftypes.JSONAnyPtr(""),
+		Location:  fftypes.JSONAnyPtr(""),
+		Method: &fftypes.FFIMethod{
+			Name:    "doStuff",
+			ID:      fftypes.NewUUID(),
+			Params:  fftypes.FFIParams{},
+			Returns: fftypes.FFIParams{},
+		},
+	}
+
+	mim.On("ResolveSigningKey", mock.Anything, "").Return("key-resolved", nil)
+	mdi.On("UpsertTransaction", mock.Anything, mock.MatchedBy(func(tx *fftypes.Transaction) bool {
+		return tx.Namespace == "ns1" && tx.Type == fftypes.TransactionTypeContractInvoke
+	})).Return(fmt.Errorf("pop"))
+
+	_, err := cm.InvokeContract(context.Background(), "ns1", req)
+
+	assert.EqualError(t, err, "pop")
+}
+
 func TestInvokeContractNoMethodSignature(t *testing.T) {
 	cm := newTestContractManager()
-	mbi := cm.blockchain.(*blockchainmocks.Plugin)
 	mim := cm.identity.(*identitymanagermocks.Manager)
 
 	req := &fftypes.ContractCallRequest{
@@ -1145,7 +1181,6 @@ func TestInvokeContractNoMethodSignature(t *testing.T) {
 	}
 
 	mim.On("ResolveSigningKey", mock.Anything, "").Return("key-resolved", nil)
-	mbi.On("InvokeContract", mock.Anything, mock.AnythingOfType("*fftypes.UUID"), "key-resolved", mock.Anything, mock.AnythingOfType("*fftypes.FFIMethod"), mock.Anything).Return(struct{}{}, nil)
 
 	_, err := cm.InvokeContract(context.Background(), "ns1", req)
 
@@ -1220,6 +1255,7 @@ func TestQueryContract(t *testing.T) {
 	cm := newTestContractManager()
 	mbi := cm.blockchain.(*blockchainmocks.Plugin)
 	mim := cm.identity.(*identitymanagermocks.Manager)
+	mdi := cm.database.(*databasemocks.Plugin)
 
 	req := &fftypes.ContractCallRequest{
 		Type:      fftypes.CallTypeQuery,
@@ -1235,6 +1271,12 @@ func TestQueryContract(t *testing.T) {
 	}
 
 	mim.On("ResolveSigningKey", mock.Anything, "").Return("key-resolved", nil)
+	mdi.On("UpsertTransaction", mock.Anything, mock.MatchedBy(func(tx *fftypes.Transaction) bool {
+		return tx.Namespace == "ns1" && tx.Type == fftypes.TransactionTypeContractInvoke
+	})).Return(nil)
+	mdi.On("InsertOperation", mock.Anything, mock.MatchedBy(func(op *fftypes.Operation) bool {
+		return op.Namespace == "ns1" && op.Type == fftypes.OpTypeContractInvoke && op.Plugin == "mockblockchain"
+	})).Return(nil)
 	mbi.On("QueryContract", mock.Anything, req.Location, req.Method, req.Input).Return(struct{}{}, nil)
 
 	_, err := cm.InvokeContract(context.Background(), "ns1", req)
@@ -1245,6 +1287,7 @@ func TestQueryContract(t *testing.T) {
 func TestCallContractInvalidType(t *testing.T) {
 	cm := newTestContractManager()
 	mim := cm.identity.(*identitymanagermocks.Manager)
+	mdi := cm.database.(*databasemocks.Plugin)
 
 	req := &fftypes.ContractCallRequest{
 		Interface: fftypes.NewUUID(),
@@ -1259,6 +1302,12 @@ func TestCallContractInvalidType(t *testing.T) {
 	}
 
 	mim.On("ResolveSigningKey", mock.Anything, "").Return("key-resolved", nil)
+	mdi.On("UpsertTransaction", mock.Anything, mock.MatchedBy(func(tx *fftypes.Transaction) bool {
+		return tx.Namespace == "ns1" && tx.Type == fftypes.TransactionTypeContractInvoke
+	})).Return(nil)
+	mdi.On("InsertOperation", mock.Anything, mock.MatchedBy(func(op *fftypes.Operation) bool {
+		return op.Namespace == "ns1" && op.Type == fftypes.OpTypeContractInvoke && op.Plugin == "mockblockchain"
+	})).Return(nil)
 
 	assert.PanicsWithValue(t, "unknown call type: ", func() {
 		cm.InvokeContract(context.Background(), "ns1", req)
@@ -1406,6 +1455,7 @@ func TestInvokeContractAPI(t *testing.T) {
 	mdb := cm.database.(*databasemocks.Plugin)
 	mim := cm.identity.(*identitymanagermocks.Manager)
 	mbi := cm.blockchain.(*blockchainmocks.Plugin)
+	mdi := cm.database.(*databasemocks.Plugin)
 
 	req := &fftypes.ContractCallRequest{
 		Type:      fftypes.CallTypeInvoke,
@@ -1427,6 +1477,12 @@ func TestInvokeContractAPI(t *testing.T) {
 	mim.On("ResolveSigningKey", mock.Anything, "").Return("key-resolved", nil)
 	mdb.On("GetContractAPIByName", mock.Anything, "ns1", "banana").Return(api, nil)
 	mdb.On("GetFFIMethod", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(&fftypes.FFIMethod{Name: "peel"}, nil)
+	mdi.On("UpsertTransaction", mock.Anything, mock.MatchedBy(func(tx *fftypes.Transaction) bool {
+		return tx.Namespace == "ns1" && tx.Type == fftypes.TransactionTypeContractInvoke
+	})).Return(nil)
+	mdi.On("InsertOperation", mock.Anything, mock.MatchedBy(func(op *fftypes.Operation) bool {
+		return op.Namespace == "ns1" && op.Type == fftypes.OpTypeContractInvoke && op.Plugin == "mockblockchain"
+	})).Return(nil)
 	mbi.On("InvokeContract", mock.Anything, mock.AnythingOfType("*fftypes.UUID"), "key-resolved", req.Location, mock.AnythingOfType("*fftypes.FFIMethod"), req.Input).Return(struct{}{}, nil)
 
 	_, err := cm.InvokeContractAPI(context.Background(), "ns1", "banana", "peel", req)
diff --git a/internal/database/sqlcommon/message_sql_test.go b/internal/database/sqlcommon/message_sql_test.go
index ab00cf656..0ff43f77c 100644
--- a/internal/database/sqlcommon/message_sql_test.go
+++ b/internal/database/sqlcommon/message_sql_test.go
@@ -61,7 +61,7 @@ func TestUpsertE2EWithDB(t *testing.T) {
 			TxType:    fftypes.TransactionTypeNone,
 		},
 		Hash:      fftypes.NewRandB32(),
-		State:     fftypes.MessageStateStaged,
+		State:     fftypes.MessageStateReady,
 		Confirmed: nil,
 		Data: []*fftypes.DataRef{
 			{ID: dataID1, Hash: rand1},
@@ -69,7 +69,7 @@ func TestUpsertE2EWithDB(t *testing.T) {
 		},
 	}
 
-	s.callbacks.On("OrderedUUIDCollectionNSEvent", database.CollectionMessages, fftypes.ChangeEventTypeCreated, "ns12345", msgID, mock.Anything).Return().Twice()
+	s.callbacks.On("OrderedUUIDCollectionNSEvent", database.CollectionMessages, fftypes.ChangeEventTypeCreated, "ns12345", msgID, mock.Anything).Return()
 	s.callbacks.On("OrderedUUIDCollectionNSEvent", database.CollectionMessages, fftypes.ChangeEventTypeUpdated, "ns12345", msgID, mock.Anything).Return()
 
 	err := s.UpsertMessage(ctx, msg, database.UpsertOptimizationNew)
@@ -196,15 +196,6 @@ func TestUpsertE2EWithDB(t *testing.T) {
 	assert.Equal(t, 1, len(msgs))
 	assert.Equal(t, *bid2, *msgs[0].BatchID)
 
-	// Bump and Update - this is for a ready transition
-	msgUpdated.State = fftypes.MessageStateReady
-	err = s.ReplaceMessage(context.Background(), msgUpdated)
-	assert.NoError(t, err)
-	msgRead, err = s.GetMessageByID(ctx, msgUpdated.Header.ID)
-	msgJson, _ = json.Marshal(&msgUpdated)
-	msgReadJson, _ = json.Marshal(msgRead)
-	assert.Equal(t, string(msgJson), string(msgReadJson))
-
 	s.callbacks.AssertExpectations(t)
 }
 
@@ -276,38 +267,6 @@ func TestUpsertMessageFailCommit(t *testing.T) {
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
 
-func TestReplaceMessageFailBegin(t *testing.T) {
-	s, mock := newMockProvider().init()
-	mock.ExpectBegin().WillReturnError(fmt.Errorf("pop"))
-	msgID := fftypes.NewUUID()
-	err := s.ReplaceMessage(context.Background(), &fftypes.Message{Header: fftypes.MessageHeader{ID: msgID}})
-	assert.Regexp(t, "FF10114", err)
-	assert.NoError(t, mock.ExpectationsWereMet())
-}
-
-func TestReplaceMessageFailDelete(t *testing.T) {
-	s, mock := newMockProvider().init()
-	mock.ExpectBegin()
-	mock.ExpectExec("DELETE .*").WillReturnError(fmt.Errorf("pop"))
-	mock.ExpectRollback()
-	msgID := fftypes.NewUUID()
-	err := s.ReplaceMessage(context.Background(), &fftypes.Message{Header: fftypes.MessageHeader{ID: msgID}})
-	assert.Regexp(t, "FF10118", err)
-	assert.NoError(t, mock.ExpectationsWereMet())
-}
-
-func TestReplaceMessageFailInsert(t *testing.T) {
-	s, mock := newMockProvider().init()
-	mock.ExpectBegin()
-	mock.ExpectExec("DELETE .*").WillReturnResult(sqlmock.NewResult(1, 1))
-	mock.ExpectExec("INSERT .*").WillReturnError(fmt.Errorf("pop"))
-	mock.ExpectRollback()
-	msgID := fftypes.NewUUID()
-	err := s.ReplaceMessage(context.Background(), &fftypes.Message{Header: fftypes.MessageHeader{ID: msgID}})
-	assert.Regexp(t, "FF10116", err)
-	assert.NoError(t, mock.ExpectationsWereMet())
-}
-
 func TestUpdateMessageDataRefsNilID(t *testing.T) {
 	s, mock := newMockProvider().init()
 	msgID := fftypes.NewUUID()
diff --git a/internal/events/operation_update_test.go b/internal/events/operation_update_test.go
index 887bc2f78..751586582 100644
--- a/internal/events/operation_update_test.go
+++ b/internal/events/operation_update_test.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2021 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -86,72 +86,48 @@ func TestOperationUpdateTransferFail(t *testing.T) {
 	mdi := em.database.(*databasemocks.Plugin)
 	mbi := &blockchainmocks.Plugin{}
 
+	opID := fftypes.NewUUID()
 	op := &fftypes.Operation{
-		ID:        fftypes.NewUUID(),
+		ID:        opID,
 		Type:      fftypes.OpTypeTokenTransfer,
 		Namespace: "ns1",
 	}
 
-	mdi.On("GetOperationByID", em.ctx, op.ID).Return(op, nil)
-	mdi.On("UpdateOperation", em.ctx, op.ID, mock.Anything).Return(nil)
-	mdi.On("UpdateTransaction", em.ctx, op.Transaction, mock.Anything).Return(nil)
+	mdi.On("GetOperationByID", em.ctx, opID).Return(op, nil)
+	mdi.On("UpdateOperation", em.ctx, opID, mock.Anything).Return(nil)
 	mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(e *fftypes.Event) bool {
 		return e.Type == fftypes.EventTypeTransferOpFailed && e.Namespace == "ns1"
 	})).Return(nil)
 
 	info := fftypes.JSONObject{"some": "info"}
-	err := em.OperationUpdate(mbi, op.ID, fftypes.OpStatusFailed, "some error", info)
+	err := em.OperationUpdate(mbi, opID, fftypes.OpStatusFailed, "some error", info)
 	assert.NoError(t, err)
 
 	mdi.AssertExpectations(t)
 	mbi.AssertExpectations(t)
 }
 
-func TestOperationUpdateTransferTransactionFail(t *testing.T) {
-	em, cancel := newTestEventManager(t)
-	defer cancel()
-	mdi := em.database.(*databasemocks.Plugin)
-	mbi := &blockchainmocks.Plugin{}
-
-	op := &fftypes.Operation{
-		ID:        fftypes.NewUUID(),
-		Type:      fftypes.OpTypeTokenTransfer,
-		Namespace: "ns1",
-	}
-
-	mdi.On("GetOperationByID", em.ctx, op.ID).Return(op, nil)
-	mdi.On("UpdateOperation", em.ctx, op.ID, mock.Anything).Return(nil)
-	mdi.On("UpdateTransaction", em.ctx, op.Transaction, mock.Anything).Return(fmt.Errorf("pop"))
-
-	info := fftypes.JSONObject{"some": "info"}
-	err := em.OperationUpdate(mbi, op.ID, fftypes.OpStatusFailed, "some error", info)
-	assert.EqualError(t, err, "pop")
-
-	mdi.AssertExpectations(t)
-	mbi.AssertExpectations(t)
-}
-
 func TestOperationUpdateTransferEventFail(t *testing.T) {
 	em, cancel := newTestEventManager(t)
 	defer cancel()
 	mdi := em.database.(*databasemocks.Plugin)
 	mbi := &blockchainmocks.Plugin{}
 
+	opID := fftypes.NewUUID()
 	op := &fftypes.Operation{
-		ID:        fftypes.NewUUID(),
+		ID:        opID,
 		Type:      fftypes.OpTypeTokenTransfer,
 		Namespace: "ns1",
 	}
 
-	mdi.On("GetOperationByID", em.ctx, op.ID).Return(op, nil)
-	mdi.On("UpdateOperation", em.ctx, op.ID, mock.Anything).Return(nil)
-	mdi.On("UpdateTransaction", em.ctx, op.Transaction, mock.Anything).Return(nil)
+	mdi.On("GetOperationByID", em.ctx, opID).Return(op, nil)
+	mdi.On("UpdateOperation", em.ctx, opID, mock.Anything).Return(nil)
 	mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(e *fftypes.Event) bool {
 		return e.Type == fftypes.EventTypeTransferOpFailed && e.Namespace == "ns1"
 	})).Return(fmt.Errorf("pop"))
 
 	info := fftypes.JSONObject{"some": "info"}
-	err := em.OperationUpdate(mbi, op.ID, fftypes.OpStatusFailed, "some error", info)
+	err := em.OperationUpdate(mbi, opID, fftypes.OpStatusFailed, "some error", info)
 	assert.EqualError(t, err, "pop")
 
 	mdi.AssertExpectations(t)
diff --git a/internal/events/tokens_transferred_test.go b/internal/events/tokens_transferred_test.go
index 4ae5aaafb..a990ab8e0 100644
--- a/internal/events/tokens_transferred_test.go
+++ b/internal/events/tokens_transferred_test.go
@@ -23,6 +23,7 @@ import (
 	"github.com/hyperledger/firefly/mocks/databasemocks"
 	"github.com/hyperledger/firefly/mocks/tokenmocks"
 	"github.com/hyperledger/firefly/pkg/blockchain"
+	"github.com/hyperledger/firefly/pkg/database"
 	"github.com/hyperledger/firefly/pkg/fftypes"
 	"github.com/hyperledger/firefly/pkg/tokens"
 	"github.com/stretchr/testify/assert"
@@ -429,9 +430,10 @@ func TestTokensTransferredWithMessageSend(t *testing.T) {
 	mdi.On("UpsertTokenTransfer", em.ctx, &transfer.TokenTransfer).Return(nil).Times(2)
 	mdi.On("UpdateTokenBalances", em.ctx, &transfer.TokenTransfer).Return(nil).Times(2)
 	mdi.On("GetMessageByID", em.ctx, mock.Anything).Return(message, nil).Times(2)
-	mdi.On("ReplaceMessage", em.ctx, mock.MatchedBy(func(msg *fftypes.Message) bool {
+	mdi.On("UpsertMessage", em.ctx, mock.Anything, database.UpsertOptimizationExisting).Return(fmt.Errorf("pop"))
+	mdi.On("UpsertMessage", em.ctx, mock.MatchedBy(func(msg *fftypes.Message) bool {
 		return msg.State == fftypes.MessageStateReady
-	})).Return(fmt.Errorf("pop"))
+	}), database.UpsertOptimizationExisting).Return(nil)
 	mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(ev *fftypes.Event) bool {
 		return ev.Type == fftypes.EventTypeTransferConfirmed && ev.Reference == transfer.LocalID && ev.Namespace == pool.Namespace
 	})).Return(nil).Once()