Merge pull request #67 from KenshiTech/v0.11.12

V0.11.12

src/constants/constants.go

@@ -1,4 +1,4 @@
 package constants
 
-var Version = "0.11.11"
+var Version = "0.11.12"
 var ProtocolVersion = "0.11.9"

src/datasets/logs.go

@@ -1,10 +1,38 @@
 package datasets
 
+import (
+	"encoding/json"
+
+	"github.com/vmihailenco/msgpack/v5"
+)
+
 type EventLogArg struct {
 	Name  string
+	Type  string
 	Value any
 }
 
+var _ msgpack.CustomEncoder = (*EventLogArg)(nil)
+
+// TODO: this can be improved
+func (eventLog *EventLogArg) EncodeMsgpack(enc *msgpack.Encoder) error {
+	encoded, err := json.Marshal(eventLog)
+	if err != nil {
+		return err
+	}
+	return enc.EncodeBytes(encoded)
+}
+
+var _ msgpack.CustomDecoder = (*EventLogArg)(nil)
+
+func (eventLog *EventLogArg) DecodeMsgpack(dec *msgpack.Decoder) error {
+	bytes, err := dec.DecodeBytes()
+	if err != nil {
+		return err
+	}
+	return json.Unmarshal(bytes, eventLog)
+}
+
 type EventLog struct {
 	LogIndex uint64
 	Block    uint64

src/gqlgen.yml

@@ -45,4 +45,8 @@ models:
       - github.com/99designs/gqlgen/graphql.Uint
       - github.com/99designs/gqlgen/graphql.Uint32
   Bytes:
-    model: github.com/KenshiTech/unchained/gql/types.Bytes
+    model:
+      - github.com/KenshiTech/unchained/gql/types.Bytes
+  Any:
+    model:
+      - github.com/99designs/gqlgen/graphql.Any

src/net/client/client.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
+	"sync"
 	"time"
 
 	"github.com/KenshiTech/unchained/bls"
@@ -20,6 +21,13 @@ import (
 var Client *websocket.Conn
 var IsClientSocketClosed = false
 var Done chan struct{}
+var mu *sync.Mutex
+
+func Send(data []byte) error {
+	mu.Lock()
+	defer mu.Unlock()
+	return Client.WriteMessage(websocket.BinaryMessage, data)
+}
 
 func StartClient() {
 
@@ -179,3 +187,7 @@ func ClientBlock() {
 		}
 	}
 }
+
+func init() {
+	mu = new(sync.Mutex)
+}

src/net/server.go

@@ -436,7 +436,7 @@ func handleAtRoot(w http.ResponseWriter, r *http.Request) {
 				messageType,
 				append(
 					[]byte{opcodes.Error},
-					[]byte("Instruction not supported")...),
+					[]byte(fmt.Sprintf("Instruction not supported: %x", payload[0]))...),
 			)
 			if err != nil {
 				fmt.Println("write:", err)

src/plugins/logs/logs.go

@@ -6,6 +6,7 @@ import (
 	"math/big"
 	"os"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 
@@ -22,7 +23,6 @@ import (
 	"github.com/KenshiTech/unchained/net/consumer"
 	"github.com/KenshiTech/unchained/persistence"
 	"github.com/KenshiTech/unchained/utils"
-	"github.com/gorilla/websocket"
 
 	bls12381 "github.com/consensys/gnark-crypto/ecc/bls12-381"
 	"github.com/dgraph-io/badger/v4"
@@ -121,7 +121,7 @@ func RecordSignature(
 
 		// TODO: this won't work for Arbitrum
 		// TODO: we disallow syncing historical events here
-		if *blockNumber-info.Block > 16 {
+		if *blockNumber-info.Block > 96 {
 			return // Data too old
 		}
 	}
@@ -324,15 +324,15 @@ func createTask(configs []LogConf, chain string) func() {
 				return
 			}
 
-			if lastSyncedBlock[conf] == allowedBlock {
+			if lastSyncedBlock[conf]+1 >= allowedBlock {
 				return
 			}
 
 			contractAddress := common.HexToAddress(conf.Address)
 			contextKey := fmt.Sprintf("plugins.logs.events.%s", conf.Name)
-			fromBlock := lastSyncedBlock[conf]
+			fromBlock := lastSyncedBlock[conf] + 1
 
-			if fromBlock == 0 {
+			if lastSyncedBlock[conf] == 0 {
 				contextBlock, err := persistence.ReadUInt64(contextKey)
 
 				if err != nil && err != badger.ErrKeyNotFound {
@@ -417,11 +417,27 @@ func createTask(configs []LogConf, chain string) func() {
 
 				message.Info(conf.Name)
 
+				argTypes := make(map[string]string)
+				for _, input := range eventAbi.Inputs {
+					argTypes[input.Name] = input.Type.String()
+				}
+
 				args := []datasets.EventLogArg{}
 				for _, key := range keys {
+
+					value := eventData[key]
+
+					if strings.HasPrefix(argTypes[key], "uint") || strings.HasPrefix(argTypes[key], "int") {
+						value = value.(*big.Int).String()
+					}
+
 					args = append(
 						args,
-						datasets.EventLogArg{Name: key, Value: eventData[key]},
+						datasets.EventLogArg{
+							Name:  key,
+							Value: value,
+							Type:  argTypes[key],
+						},
 					)
 				}
 
@@ -456,8 +472,7 @@ func createTask(configs []LogConf, chain string) func() {
 				}
 
 				if conf.Send {
-					client.Client.WriteMessage(
-						websocket.BinaryMessage,
+					client.Send(
 						append([]byte{opcodes.EventLog, 0}, payload...),
 					)
 				}
@@ -571,19 +586,19 @@ func init() {
 	supportedEvents = make(map[SupportKey]bool)
 
 	var err error
-	signatureCache, err = lru.New[bls12381.G1Affine, []bls.Signature](24)
+	signatureCache, err = lru.New[bls12381.G1Affine, []bls.Signature](128)
 
 	if err != nil {
 		panic(err)
 	}
 
-	consensus, err = lru.New[EventKey, map[bls12381.G1Affine]uint64](24)
+	consensus, err = lru.New[EventKey, map[bls12381.G1Affine]uint64](128)
 
 	if err != nil {
 		panic(err)
 	}
 
-	aggregateCache, err = lru.New[bls12381.G1Affine, bls12381.G1Affine](24)
+	aggregateCache, err = lru.New[bls12381.G1Affine, bls12381.G1Affine](128)
 
 	if err != nil {
 		panic(err)

src/plugins/uniswap/uniswap.go

@@ -24,7 +24,6 @@ import (
 
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/go-co-op/gocron/v2"
-	"github.com/gorilla/websocket"
 	"github.com/puzpuzpuz/xsync/v3"
 	"github.com/vmihailenco/msgpack/v5"
 	"golang.org/x/text/cases"
@@ -494,8 +493,7 @@ func syncBlocks(token Token, key TokenKey, latest uint64) {
 		}
 
 		if token.Send && !client.IsClientSocketClosed {
-			client.Client.WriteMessage(
-				websocket.BinaryMessage,
+			client.Send(
 				append([]byte{opcodes.PriceReport, 0}, payload...),
 			)
 		}