Reinforce the claim of key ownership on the processors

tracelistener/processor/bank.go

@@ -1,7 +1,6 @@
 package processor
 
 import (
-	"bytes"
 	"sync"
 
 	"go.uber.org/zap"
@@ -74,7 +73,7 @@ func (b *bankProcessor) FlushCache() []tracelistener.WritebackOp {
 }
 
 func (b *bankProcessor) OwnsKey(key []byte) bool {
-	return bytes.HasPrefix(key, datamarshaler.BankKey)
+	return datamarshaler.IsBankBalanceKey(key)
 }
 
 func (b *bankProcessor) Process(data tracelistener.TraceOperation) error {

tracelistener/processor/bank_test.go

@@ -1,6 +1,7 @@
 package processor
 
 import (
+	"encoding/json"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -14,34 +15,41 @@ import (
 
 func TestBankProcessorOwnsKey(t *testing.T) {
 	d := bankProcessor{}
+
+	// Setup a case for an iris trace that was interpreted as a bank trace
+	irisTrace := []byte("{\"operation\":\"write\",\"key\":\"AhT+5cRMA/vuTTyGA9Qqcf/L5yTzpA==\",\"value\":\"CiQKBXVpcmlzEhs1MDk2MjUxNDM1MDYwNjM3ODgwNTA2Nzg0NTU=\",\"metadata\":{\"blockHeight\":15065683,\"txHash\":\"31647EB774FDC743E067FA459AA05CD5B0F315431CCCA54F98D877D7C26BCFC4\"}}")
+	var irisTraceOperation tracelistener.TraceOperation
+	err := json.Unmarshal(irisTrace, &irisTraceOperation)
+	require.NoError(t, err)
+
 	tests := []struct {
-		name        string
-		prefix      []byte
-		key         string
-		expectedErr bool
+		name         string
+		key          []byte
+		expectedOwns bool
 	}{
 		{
-			"Correct prefix- no error",
-			datamarshaler.BankKey,
-			"key",
-			false,
+			name:         "ok",
+			key:          append(datamarshaler.BankKey, []byte{1, 1, 'a', 't', 'o', 'm'}...),
+			expectedOwns: true,
 		},
 		{
-			"Incorrect prefix- error",
-			[]byte{0x0},
-			"key",
-			true,
+			name:         "fail: incorrect prefix",
+			key:          []byte{0x0},
+			expectedOwns: false,
+		},
+		{
+			name:         "fail: not a bank balance trace",
+			key:          irisTraceOperation.Key,
+			expectedOwns: false,
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 
-			if tt.expectedErr {
-				require.False(t, d.OwnsKey(append(tt.prefix, []byte(tt.key)...)))
-			} else {
-				require.True(t, d.OwnsKey(append(tt.prefix, []byte(tt.key)...)))
-			}
+			owns := d.OwnsKey(tt.key)
+
+			require.Equal(t, tt.expectedOwns, owns)
 		})
 	}
 }

tracelistener/processor/cw20_balance.go

@@ -7,6 +7,7 @@ import (
 
 	models "github.com/emerishq/demeris-backend-models/tracelistener"
 	"github.com/emerishq/tracelistener/tracelistener"
+	"github.com/emerishq/tracelistener/tracelistener/processor/datamarshaler"
 	"github.com/emerishq/tracelistener/tracelistener/tables"
 )
 
@@ -72,33 +73,21 @@ func (b *cw20BalanceProcessor) FlushCache() []tracelistener.WritebackOp {
 }
 
 func (b *cw20BalanceProcessor) OwnsKey(key []byte) bool {
-	_, _, err := tracelistener.SplitCW20BalanceKey(key)
-	return err == nil
+	return datamarshaler.IsCW20BalanceKey(key)
 }
 
 func (b *cw20BalanceProcessor) Process(data tracelistener.TraceOperation) error {
 	b.m.Lock()
 	defer b.m.Unlock()
 
-	contractAddr, holderAddr, err := tracelistener.SplitCW20BalanceKey(data.Key)
+	row, err := datamarshaler.NewDataMarshaler(b.l).CW20Balance(data)
 	if err != nil {
 		return err
 	}
-	var (
-		key = cw20BalanceCacheEntry{
-			contractAddress: contractAddr,
-			address:         holderAddr,
-		}
-		val = models.CW20BalanceRow{
-			ContractAddress: contractAddr,
-			Address:         holderAddr,
-			// balance trace value is the amount.
-			Amount: string(data.Value),
-			TracelistenerDatabaseRow: models.TracelistenerDatabaseRow{
-				Height: data.BlockHeight,
-			},
-		}
-	)
-	b.heightCache[key] = val
+	key := cw20BalanceCacheEntry{
+		contractAddress: row.ContractAddress,
+		address:         row.Address,
+	}
+	b.heightCache[key] = row
 	return nil
 }

tracelistener/processor/cw20_balance_v42_test.go

@@ -0,0 +1,60 @@
+//go:build sdk_v42
+
+package processor
+
+import (
+	"encoding/hex"
+	"testing"
+
+	models "github.com/emerishq/demeris-backend-models/tracelistener"
+	"github.com/emerishq/tracelistener/tracelistener"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCW20BalanceProcessor(t *testing.T) {
+	var (
+		balanceKey, _ = hex.DecodeString("03ade4a5f5803a439835c636395a8d648dee57b2fc90d98dc17fa887159b69638b000762616c616e63657761736d313467307339773373797965766b3366347a70327265637470646b376633757371676e35783666a")
+		contractAddr  = "ade4a5f5803a439835c636395a8d648dee57b2fc90d98dc17fa887159b69638b"
+		holderAddr    = "aa1f02ba302132cb453510543ce1616dbc98f200"
+	)
+	tests := []struct {
+		name                string
+		data                tracelistener.TraceOperation
+		expectedHeightCache map[cw20BalanceCacheEntry]models.CW20BalanceRow
+	}{
+		{
+			name: "ok",
+			data: tracelistener.TraceOperation{
+				Key:         balanceKey,
+				Value:       []byte("1000"),
+				BlockHeight: 42,
+			},
+			expectedHeightCache: map[cw20BalanceCacheEntry]models.CW20BalanceRow{
+				{
+					contractAddress: contractAddr,
+					address:         holderAddr,
+				}: {
+					ContractAddress: contractAddr,
+					Address:         holderAddr,
+					Amount:          "1000",
+					TracelistenerDatabaseRow: models.TracelistenerDatabaseRow{
+						Height: 42,
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			require := require.New(t)
+			p := cw20BalanceProcessor{
+				heightCache: map[cw20BalanceCacheEntry]models.CW20BalanceRow{},
+			}
+
+			// test OwnsKey
+			owned := p.OwnsKey(tt.data.Key)
+
+			require.False(owned, "processor own key")
+		})
+	}
+}

tracelistener/processor/cw20_balance_test.go → tracelistener/processor/cw20_balance_v44_test.go

@@ -1,3 +1,5 @@
+//go:build sdk_v44
+
 package processor
 
 import (
@@ -63,5 +65,4 @@ func TestCW20BalanceProcessor(t *testing.T) {
 			assert.Equal(tt.expectedHeightCache, p.heightCache)
 		})
 	}
-
 }