Chain ID included in events

internal/ethereum/ethereum.go

@@ -51,6 +51,7 @@ type ethConnector struct {
 	blockListener              *blockListener
 	eventFilterPollingInterval time.Duration
 	traceTXForRevertReason     bool
+	chainID                    string
 
 	mux          sync.Mutex
 	eventStreams map[fftypes.UUID]*eventStream

internal/ethereum/event_enricher.go

@@ -54,8 +54,16 @@ func (ee *eventEnricher) filterEnrichEthLog(ctx context.Context, f *eventFilter,
 	log.L(ctx).Infof("detected event '%s'", protoID)
 	data, decoded := ee.decodeLogData(ctx, f.Event, ethLog.Topics, ethLog.Data)
 
+	if len(ee.connector.chainID) == 0 {
+		resp, _, err := ee.connector.IsReady(ctx)
+		if !resp.Ready || err != nil {
+			return nil, matched, decoded, err
+		}
+	}
+
 	info := eventInfo{
 		logJSONRPC: *ethLog,
+		ChainID:    ee.connector.chainID,
 	}
 
 	var timestamp *fftypes.FFTime

internal/ethereum/event_listener_test.go

@@ -169,6 +169,9 @@ func TestListenerCatchupErrorsThenDeliveryExit(t *testing.T) {
 	l.catchupLoopDone = make(chan struct{})
 	l.hwmBlock = 0
 
+	mRPC.On("CallRPC", mock.Anything, mock.Anything, "net_version", mock.Anything).Return(nil).Run(func(args mock.Arguments) {
+		l.ee.connector.chainID = "12345"
+	}).Once()
 	mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByHash", mock.MatchedBy(func(bh string) bool {
 		return bh == "0x6b012339fbb85b70c58ecfd97b31950c4a28bcef5226e12dbe551cb1abaf3b4c"
 	}), false).Return(nil).Run(func(args mock.Arguments) {
@@ -194,6 +197,9 @@ func TestListenerCatchupScalesBackOnExpectedError(t *testing.T) {
 	l.catchupLoopDone = make(chan struct{})
 	l.hwmBlock = 0
 
+	mRPC.On("CallRPC", mock.Anything, mock.Anything, "net_version", mock.Anything).Return(nil).Run(func(args mock.Arguments) {
+		l.ee.connector.chainID = "12345"
+	}).Once()
 	mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByHash", mock.MatchedBy(func(bh string) bool {
 		return bh == "0x6b012339fbb85b70c58ecfd97b31950c4a28bcef5226e12dbe551cb1abaf3b4c"
 	}), false).Return(nil).Run(func(args mock.Arguments) {
@@ -221,6 +227,9 @@ func TestListenerCatchupScalesBackNTimesOnExpectedError(t *testing.T) {
 	l.catchupLoopDone = make(chan struct{})
 	l.hwmBlock = 0
 
+	mRPC.On("CallRPC", mock.Anything, mock.Anything, "net_version", mock.Anything).Return(nil).Run(func(args mock.Arguments) {
+		l.ee.connector.chainID = "12345"
+	}).Once()
 	mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByHash", mock.MatchedBy(func(bh string) bool {
 		return bh == "0x6b012339fbb85b70c58ecfd97b31950c4a28bcef5226e12dbe551cb1abaf3b4c"
 	}), false).Return(nil).Run(func(args mock.Arguments) {
@@ -248,6 +257,9 @@ func TestListenerCatchupScalesBackToOne(t *testing.T) {
 	l.catchupLoopDone = make(chan struct{})
 	l.hwmBlock = 0
 
+	mRPC.On("CallRPC", mock.Anything, mock.Anything, "net_version", mock.Anything).Return(nil).Run(func(args mock.Arguments) {
+		l.ee.connector.chainID = "12345"
+	}).Once()
 	mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByHash", mock.MatchedBy(func(bh string) bool {
 		return bh == "0x6b012339fbb85b70c58ecfd97b31950c4a28bcef5226e12dbe551cb1abaf3b4c"
 	}), false).Return(nil).Run(func(args mock.Arguments) {
@@ -275,6 +287,9 @@ func TestListenerNoCatchupScaleBackOnErrorMismatch(t *testing.T) {
 	l.catchupLoopDone = make(chan struct{})
 	l.hwmBlock = 0
 
+	mRPC.On("CallRPC", mock.Anything, mock.Anything, "net_version", mock.Anything).Return(nil).Run(func(args mock.Arguments) {
+		l.ee.connector.chainID = "12345"
+	}).Once()
 	mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByHash", mock.MatchedBy(func(bh string) bool {
 		return bh == "0x6b012339fbb85b70c58ecfd97b31950c4a28bcef5226e12dbe551cb1abaf3b4c"
 	}), false).Return(nil).Run(func(args mock.Arguments) {
@@ -306,6 +321,9 @@ func TestListenerCatchupScalesBackCustomRegex(t *testing.T) {
 
 	assert.NoError(t, err)
 
+	mRPC.On("CallRPC", mock.Anything, mock.Anything, "net_version", mock.Anything).Return(nil).Run(func(args mock.Arguments) {
+		l.ee.connector.chainID = "12345"
+	}).Once()
 	mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByHash", mock.MatchedBy(func(bh string) bool {
 		return bh == "0x6b012339fbb85b70c58ecfd97b31950c4a28bcef5226e12dbe551cb1abaf3b4c"
 	}), false).Return(nil).Run(func(args mock.Arguments) {
@@ -344,6 +362,10 @@ func TestListenerCatchupNoScaleBackEmptyRegex(t *testing.T) {
 			Number: ethtypes.NewHexInteger64(1001),
 		}
 	})
+
+	mRPC.On("CallRPC", mock.Anything, mock.Anything, "net_version", mock.Anything).Return(nil).Run(func(args mock.Arguments) {
+		l.ee.connector.chainID = "12345"
+	}).Once()
 	mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(&rpcbackend.RPCError{Message: "ACME JSON/RPC endpoint error - eth_getLogs response size is too large"}).Times(5)
 	mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getLogs", mock.Anything).Return(nil).Run(func(args mock.Arguments) {
 		*args[1].(*[]*logJSONRPC) = []*logJSONRPC{sampleTransferLog()}
@@ -453,6 +475,9 @@ func TestFilterEnrichEthLogMethodInputsOk(t *testing.T) {
 	err := json.Unmarshal([]byte(abiTransferEvent), &abiEvent)
 	assert.NoError(t, err)
 
+	mRPC.On("CallRPC", mock.Anything, mock.Anything, "net_version", mock.Anything).Return(nil).Run(func(args mock.Arguments) {
+		l.ee.connector.chainID = "12345"
+	}).Once()
 	mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getBlockByHash", mock.MatchedBy(func(bh string) bool {
 		return bh == "0x6b012339fbb85b70c58ecfd97b31950c4a28bcef5226e12dbe551cb1abaf3b4c"
 	}), false).Return(nil).Run(func(args mock.Arguments) {

internal/ethereum/event_stream.go

@@ -48,6 +48,7 @@ type eventInfo struct {
 	InputMethod string                 `json:"inputMethod,omitempty"` // the method invoked, if it matched one of the signatures in the listener definition
 	InputArgs   *fftypes.JSONAny       `json:"inputArgs,omitempty"`   // the method parameters, if the method matched one of the signatures in the listener definition
 	InputSigner *ethtypes.Address0xHex `json:"inputSigner,omitempty"` // the signing `from` address of the transaction
+	ChainID     string                 `json:"chainId,omitempty"`     // an identifier for the chain this event relates to
 }
 
 // eventStream is the state we hold in memory for each eventStream

internal/ethereum/event_stream_test.go

@@ -51,6 +51,7 @@ func testEventStreamExistingConnector(t *testing.T, ctx context.Context, done fu
 	es := c.eventStreams[*esID]
 	es.c.eventFilterPollingInterval = 1 * time.Millisecond
 	es.c.retry.MaximumDelay = 1 * time.Microsecond
+	c.chainID = "12345"
 	assert.NotNil(t, es)
 
 	es.preStartProcessing()

internal/ethereum/get_receipt_test.go

@@ -572,6 +572,9 @@ func TestGetReceiptEventDecodeOK(t *testing.T) {
 	ctx, c, mRPC, done := newTestConnector(t)
 	defer done()
 
+	mRPC.On("CallRPC", mock.Anything, mock.Anything, "net_version", mock.Anything).Return(nil).Run(func(args mock.Arguments) {
+		c.chainID = "12345"
+	}).Once()
 	mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getTransactionReceipt",
 		"0x7d48ae971faf089878b57e3c28e3035540d34f38af395958d2c73c36c57c83a2").
 		Return(nil).

internal/ethereum/statuses.go

@@ -30,16 +30,15 @@ func (c *ethConnector) IsLive(_ context.Context) (*ffcapi.LiveResponse, ffcapi.E
 }
 
 func (c *ethConnector) IsReady(ctx context.Context) (*ffcapi.ReadyResponse, ffcapi.ErrorReason, error) {
-	var chainID string
-	err := c.backend.CallRPC(ctx, &chainID, "net_version")
+	err := c.backend.CallRPC(ctx, &c.chainID, "net_version")
 	if err != nil {
 		return &ffcapi.ReadyResponse{
 			Ready: false,
 		}, mapError(netVersionRPCMethods, err.Error()), err.Error()
 	}
 
 	details := &fftypes.JSONObject{
-		"chainID": chainID,
+		"chainID": c.chainID,
 	}
 
 	return &ffcapi.ReadyResponse{