Merge pull request #448 from kaleido-io/blockchain-txids

Add Blockchain Transaction IDs to FireFly Transaction objects and add blockchain IDs to events
hyperledger · Jan 27, 2022 · 5526967 · 5526967
2 parents e3993c3 + 2636cfb
commit 5526967
Show file tree

Hide file tree

Showing 51 changed files with 705 additions and 454 deletions.
diff --git a/...000056_drop_transactions_columns.down.sql → ...56_refactor_transactions_columns.down.sql b/...000056_drop_transactions_columns.down.sql → ...56_refactor_transactions_columns.down.sql
@@ -7,4 +7,7 @@ ALTER TABLE transactions ADD COLUMN info        BYTEA;
 
 CREATE INDEX transactions_protocol_id ON transactions(protocol_id);
 CREATE INDEX transactions_ref ON transactions(ref);
+
+DROP INDEX transactions_blockchain_ids;
+ALTER TABLE transactions DROP COLUMN blockchain_ids;
 COMMIT;
diff --git a/...s/000056_drop_transactions_columns.up.sql → ...0056_refactor_transactions_columns.up.sql b/...s/000056_drop_transactions_columns.up.sql → ...0056_refactor_transactions_columns.up.sql
@@ -7,4 +7,7 @@ ALTER TABLE transactions DROP COLUMN signer;
 ALTER TABLE transactions DROP COLUMN hash;
 ALTER TABLE transactions DROP COLUMN protocol_id;
 ALTER TABLE transactions DROP COLUMN info;
+
+ALTER TABLE transactions ADD COLUMN blockchain_ids VARCHAR(1024);
+CREATE INDEX transactions_blockchain_ids ON transactions(blockchain_ids);
 COMMIT;
diff --git a/...000056_drop_transactions_columns.down.sql → ...56_refactor_transactions_columns.down.sql b/...000056_drop_transactions_columns.down.sql → ...56_refactor_transactions_columns.down.sql
@@ -6,3 +6,6 @@ ALTER TABLE transactions ADD COLUMN info        BYTEA;
 
 CREATE INDEX transactions_protocol_id ON transactions(protocol_id);
 CREATE INDEX transactions_ref ON transactions(ref);
+
+DROP INDEX transactions_blockchain_ids;
+ALTER TABLE transactions DROP COLUMN blockchain_ids;
diff --git a/...e/000056_drop_transactions_columns.up.sql → ...0056_refactor_transactions_columns.up.sql b/...e/000056_drop_transactions_columns.up.sql → ...0056_refactor_transactions_columns.up.sql
@@ -6,3 +6,6 @@ ALTER TABLE transactions DROP COLUMN signer;
 ALTER TABLE transactions DROP COLUMN hash;
 ALTER TABLE transactions DROP COLUMN protocol_id;
 ALTER TABLE transactions DROP COLUMN info;
+
+ALTER TABLE transactions ADD COLUMN blockchain_ids VARCHAR(1024);
+CREATE INDEX transactions_blockchain_ids ON transactions(blockchain_ids);
diff --git a/docs/swagger/swagger.yaml b/docs/swagger/swagger.yaml
@@ -5078,6 +5078,10 @@ paths:
             application/json:
               schema:
                 properties:
+                  blockchainIds:
+                    items:
+                      type: string
+                    type: array
                   created: {}
                   id: {}
                   namespace:
@@ -9837,6 +9841,11 @@ paths:
         schema:
           default: 120s
           type: string
+      - description: 'Data filter field. Prefixes supported: > >= < <= @ ^ ! !@ !^'
+        in: query
+        name: blockchainids
+        schema:
+          type: string
       - description: 'Data filter field. Prefixes supported: > >= < <= @ ^ ! !@ !^'
         in: query
         name: created
@@ -9902,6 +9911,10 @@ paths:
             application/json:
               schema:
                 properties:
+                  blockchainIds:
+                    items:
+                      type: string
+                    type: array
                   created: {}
                   id: {}
                   namespace:
@@ -9945,6 +9958,11 @@ paths:
         schema:
           default: 120s
           type: string
+      - description: 'Data filter field. Prefixes supported: > >= < <= @ ^ ! !@ !^'
+        in: query
+        name: blockchainids
+        schema:
+          type: string
       - description: 'Data filter field. Prefixes supported: > >= < <= @ ^ ! !@ !^'
         in: query
         name: created
@@ -10010,6 +10028,10 @@ paths:
             application/json:
               schema:
                 properties:
+                  blockchainIds:
+                    items:
+                      type: string
+                    type: array
                   created: {}
                   id: {}
                   namespace:

diff --git a/internal/batch/batch_processor_test.go b/internal/batch/batch_processor_test.go
@@ -294,7 +294,7 @@ func TestCalcPinsFail(t *testing.T) {
 			Messages: []*fftypes.Message{
 				{Header: fftypes.MessageHeader{
 					Group:  gid,
-					Topics: fftypes.FFNameArray{"topic1"},
+					Topics: fftypes.FFStringArray{"topic1"},
 				}},
 			},
 		},

diff --git a/internal/blockchain/ethereum/ethereum.go b/internal/blockchain/ethereum/ethereum.go
@@ -188,8 +188,10 @@ func ethHexFormatB32(b *fftypes.Bytes32) string {
 
 func (e *Ethereum) handleBatchPinEvent(ctx context.Context, msgJSON fftypes.JSONObject) (err error) {
 	sBlockNumber := msgJSON.GetString("blockNumber")
-	sTransactionIndex := msgJSON.GetString("transactionIndex")
 	sTransactionHash := msgJSON.GetString("transactionHash")
+	blockNumber := msgJSON.GetInt64("blockNumber")
+	txIndex := msgJSON.GetInt64("transactionIndex")
+	logIndex := msgJSON.GetInt64("logIndex")
 	dataJSON := msgJSON.GetObject("data")
 	authorAddress := dataJSON.GetString("author")
 	ns := dataJSON.GetString("namespace")
@@ -205,7 +207,6 @@ func (e *Ethereum) handleBatchPinEvent(ctx context.Context, msgJSON fftypes.JSON
 	}
 
 	if sBlockNumber == "" ||
-		sTransactionIndex == "" ||
 		sTransactionHash == "" ||
 		authorAddress == "" ||
 		sUUIDs == "" ||
@@ -257,12 +258,13 @@ func (e *Ethereum) handleBatchPinEvent(ctx context.Context, msgJSON fftypes.JSON
 		BatchPayloadRef: sPayloadRef,
 		Contexts:        contexts,
 		Event: blockchain.Event{
-			Source:     e.Name(),
-			Name:       "BatchPin",
-			ProtocolID: sTransactionHash,
-			Output:     dataJSON,
-			Info:       msgJSON,
-			Timestamp:  timestamp,
+			BlockchainTXID: sTransactionHash,
+			Source:         e.Name(),
+			Name:           "BatchPin",
+			ProtocolID:     fmt.Sprintf("%.12d/%.6d/%.6d", blockNumber, txIndex, logIndex),
+			Output:         dataJSON,
+			Info:           msgJSON,
+			Timestamp:      timestamp,
 		},
 	}
 
@@ -272,6 +274,9 @@ func (e *Ethereum) handleBatchPinEvent(ctx context.Context, msgJSON fftypes.JSON
 
 func (e *Ethereum) handleContractEvent(ctx context.Context, msgJSON fftypes.JSONObject) (err error) {
 	sTransactionHash := msgJSON.GetString("transactionHash")
+	blockNumber := msgJSON.GetInt64("blockNumber")
+	txIndex := msgJSON.GetInt64("transactionIndex")
+	logIndex := msgJSON.GetInt64("logIndex")
 	sub := msgJSON.GetString("subId")
 	signature := msgJSON.GetString("signature")
 	dataJSON := msgJSON.GetObject("data")
@@ -284,19 +289,20 @@ func (e *Ethereum) handleContractEvent(ctx context.Context, msgJSON fftypes.JSON
 	}
 	delete(msgJSON, "data")
 
-	event := &blockchain.ContractEvent{
+	event := &blockchain.EventWithSubscription{
 		Subscription: sub,
 		Event: blockchain.Event{
-			Source:     e.Name(),
-			Name:       name,
-			ProtocolID: sTransactionHash,
-			Output:     dataJSON,
-			Info:       msgJSON,
-			Timestamp:  timestamp,
+			BlockchainTXID: sTransactionHash,
+			Source:         e.Name(),
+			Name:           name,
+			ProtocolID:     fmt.Sprintf("%.12d/%.6d/%.6d", blockNumber, txIndex, logIndex),
+			Output:         dataJSON,
+			Info:           msgJSON,
+			Timestamp:      timestamp,
 		},
 	}
 
-	return e.callbacks.ContractEvent(event)
+	return e.callbacks.BlockchainEvent(event)
 }
 
 func (e *Ethereum) handleReceipt(ctx context.Context, reply fftypes.JSONObject) error {
@@ -321,7 +327,7 @@ func (e *Ethereum) handleReceipt(ctx context.Context, reply fftypes.JSONObject)
 		updateType = fftypes.OpStatusFailed
 	}
 	l.Infof("Ethconnect '%s' reply: request=%s tx=%s message=%s", replyType, requestID, txHash, message)
-	return e.callbacks.BlockchainOpUpdate(operationID, updateType, message, reply)
+	return e.callbacks.BlockchainOpUpdate(operationID, updateType, txHash, message, reply)
 }
 
 func (e *Ethereum) handleMessageBatch(ctx context.Context, messages []interface{}) error {

diff --git a/internal/blockchain/ethereum/ethereum_test.go b/internal/blockchain/ethereum/ethereum_test.go
@@ -983,6 +983,7 @@ func TestHandleReceiptTXSuccess(t *testing.T) {
 	em.On("BlockchainOpUpdate",
 		operationID,
 		fftypes.OpStatusSucceeded,
+		"0x71a38acb7a5d4a970854f6d638ceb1fa10a4b59cbf4ed7674273a1a8dc8b36b8",
 		"",
 		mock.Anything).Return(nil)
 
@@ -1022,6 +1023,7 @@ func TestHandleBadPayloadsAndThenReceiptFailure(t *testing.T) {
 	txsu := em.On("BlockchainOpUpdate",
 		operationID,
 		fftypes.OpStatusFailed,
+		"",
 		"Packing arguments for method 'broadcastBatch': abi: cannot use [3]uint8 as type [32]uint8 as argument",
 		mock.Anything).Return(fmt.Errorf("Shutdown"))
 	done := make(chan struct{})
@@ -1292,15 +1294,19 @@ func TestHandleMessageContractEvent(t *testing.T) {
 		ID: "sb-b5b97a4e-a317-4053-6400-1474650efcb5",
 	}
 
-	em.On("ContractEvent", mock.Anything).Return(nil)
+	em.On("BlockchainEvent", mock.MatchedBy(func(e *blockchain.EventWithSubscription) bool {
+		assert.Equal(t, "0xc26df2bf1a733e9249372d61eb11bd8662d26c8129df76890b1beb2f6fa72628", e.BlockchainTXID)
+		assert.Equal(t, "000000038011/000000/000050", e.Event.ProtocolID)
+		return true
+	})).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)
 
-	ev := em.Calls[0].Arguments[0].(*blockchain.ContractEvent)
+	ev := em.Calls[0].Arguments[0].(*blockchain.EventWithSubscription)
 	assert.Equal(t, "sub2", ev.Subscription)
 	assert.Equal(t, "Changed", ev.Event.Name)
 
@@ -1351,7 +1357,7 @@ func TestHandleMessageContractEventNoTimestamp(t *testing.T) {
 		ID: "sb-b5b97a4e-a317-4053-6400-1474650efcb5",
 	}
 
-	em.On("ContractEvent", mock.Anything).Return(nil)
+	em.On("BlockchainEvent", mock.Anything).Return(nil)
 
 	var events []interface{}
 	err := json.Unmarshal(data.Bytes(), &events)
@@ -1387,7 +1393,7 @@ func TestHandleMessageContractEventError(t *testing.T) {
 		ID: "sb-b5b97a4e-a317-4053-6400-1474650efcb5",
 	}
 
-	em.On("ContractEvent", mock.Anything).Return(fmt.Errorf("pop"))
+	em.On("BlockchainEvent", mock.Anything).Return(fmt.Errorf("pop"))
 
 	var events []interface{}
 	err := json.Unmarshal(data.Bytes(), &events)

diff --git a/internal/blockchain/ethereum/ffi_param_validator_test.go b/internal/blockchain/ethereum/ffi_param_validator_test.go
@@ -108,7 +108,7 @@ func TestSchemaTypeInvalidFFIType(t *testing.T) {
 
 func TestSchemaTypeMissing(t *testing.T) {
 	_, err := NewTestSchema(`{}`)
-	assert.Regexp(t, "missing properties: 'type'", err)
+	assert.Regexp(t, "missing properties", err)
 }
 
 func TestSchemaDetailsTypeMissing(t *testing.T) {

diff --git a/internal/blockchain/fabric/fabric.go b/internal/blockchain/fabric/fabric.go
@@ -265,6 +265,9 @@ func (f *Fabric) handleBatchPinEvent(ctx context.Context, msgJSON fftypes.JSONOb
 	}
 
 	sTransactionHash := msgJSON.GetString("transactionId")
+	blockNumber := msgJSON.GetInt64("blockNumber")
+	transactionIndex := msgJSON.GetInt64("transactionIndex")
+	eventIndex := msgJSON.GetInt64("eventIndex")
 	signer := payload.GetString("signer")
 	ns := payload.GetString("namespace")
 	sUUIDs := payload.GetString("uuids")
@@ -315,12 +318,13 @@ func (f *Fabric) handleBatchPinEvent(ctx context.Context, msgJSON fftypes.JSONOb
 		BatchPayloadRef: sPayloadRef,
 		Contexts:        contexts,
 		Event: blockchain.Event{
-			Source:     f.Name(),
-			Name:       "BatchPin",
-			ProtocolID: sTransactionHash,
-			Output:     *payload,
-			Info:       msgJSON,
-			Timestamp:  fftypes.UnixTime(timestamp),
+			BlockchainTXID: sTransactionHash,
+			Source:         f.Name(),
+			Name:           "BatchPin",
+			ProtocolID:     fmt.Sprintf("%.12d/%.6d/%.6d", blockNumber, transactionIndex, eventIndex),
+			Output:         *payload,
+			Info:           msgJSON,
+			Timestamp:      fftypes.UnixTime(timestamp),
 		},
 	}
 
@@ -337,6 +341,9 @@ func (f *Fabric) handleContractEvent(ctx context.Context, msgJSON fftypes.JSONOb
 	delete(msgJSON, "payload")
 
 	sTransactionHash := msgJSON.GetString("transactionId")
+	blockNumber := msgJSON.GetInt64("blockNumber")
+	transactionIndex := msgJSON.GetInt64("transactionIndex")
+	eventIndex := msgJSON.GetInt64("eventIndex")
 	sub := msgJSON.GetString("subId")
 	name := msgJSON.GetString("eventName")
 	sTimestamp := msgJSON.GetString("timestamp")
@@ -346,19 +353,20 @@ func (f *Fabric) handleContractEvent(ctx context.Context, msgJSON fftypes.JSONOb
 		// Continue with zero timestamp
 	}
 
-	event := &blockchain.ContractEvent{
+	event := &blockchain.EventWithSubscription{
 		Subscription: sub,
 		Event: blockchain.Event{
-			Source:     f.Name(),
-			Name:       name,
-			ProtocolID: sTransactionHash,
-			Output:     *payload,
-			Info:       msgJSON,
-			Timestamp:  fftypes.UnixTime(timestamp),
+			BlockchainTXID: sTransactionHash,
+			Source:         f.Name(),
+			Name:           name,
+			ProtocolID:     fmt.Sprintf("%.12d/%.6d/%.6d", blockNumber, transactionIndex, eventIndex),
+			Output:         *payload,
+			Info:           msgJSON,
+			Timestamp:      fftypes.UnixTime(timestamp),
 		},
 	}
 
-	return f.callbacks.ContractEvent(event)
+	return f.callbacks.BlockchainEvent(event)
 }
 
 func (f *Fabric) handleReceipt(ctx context.Context, reply fftypes.JSONObject) error {
@@ -367,7 +375,7 @@ func (f *Fabric) handleReceipt(ctx context.Context, reply fftypes.JSONObject) er
 	headers := reply.GetObject("headers")
 	requestID := headers.GetString("requestId")
 	replyType := headers.GetString("type")
-	txHash := reply.GetString("transactionHash")
+	txHash := reply.GetString("transactionId")
 	message := reply.GetString("errorMessage")
 	if requestID == "" || replyType == "" {
 		l.Errorf("Reply cannot be processed: %+v", reply)
@@ -383,7 +391,7 @@ func (f *Fabric) handleReceipt(ctx context.Context, reply fftypes.JSONObject) er
 		updateType = fftypes.OpStatusFailed
 	}
 	l.Infof("Fabconnect '%s' reply tx=%s (request=%s) %s", replyType, txHash, requestID, message)
-	return f.callbacks.BlockchainOpUpdate(operationID, updateType, message, reply)
+	return f.callbacks.BlockchainOpUpdate(operationID, updateType, txHash, message, reply)
 }
 
 func (f *Fabric) handleMessageBatch(ctx context.Context, messages []interface{}) error {