feat(driver): host the driver v3/v4 streaming controller in sentio-core

MODULE.bazel

@@ -174,6 +174,7 @@ use_repo(
     "com_github_vmihailenco_msgpack_v5",
     "com_github_wasmerio_wasmer_go",
     "com_google_cloud_go_bigquery",
+    "com_google_cloud_go_pubsub",
     "in_gopkg_natefinch_lumberjack_v2",
     "in_gopkg_yaml_v2",
     "in_gopkg_yaml_v3",
@@ -199,6 +200,7 @@ use_repo(
     "org_golang_google_protobuf",
     "org_golang_x_exp",
     "org_golang_x_net",
+    "org_golang_x_sync",
     "org_modernc_mathutil",
     "org_uber_go_mock",
     "org_uber_go_zap",

common/compress/BUILD.bazel

@@ -0,0 +1,16 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "compress",
+    srcs = ["compress.go"],
+    importpath = "sentioxyz/sentio-core/common/compress",
+    visibility = ["//visibility:public"],
+    deps = ["@com_github_pkg_errors//:errors"],
+)
+
+go_test(
+    name = "compress_test",
+    srcs = ["compress_test.go"],
+    embed = [":compress"],
+    deps = ["@com_github_stretchr_testify//assert"],
+)

common/compress/compress.go

@@ -0,0 +1,71 @@
+package compress
+
+import (
+	"bytes"
+	"compress/gzip"
+	"encoding/json"
+	"github.com/pkg/errors"
+	"io"
+)
+
+type compressPayload struct {
+	CompressMethod string `json:"compress_method,omitempty"`
+	Data           []byte `json:"data,omitempty"`
+}
+
+const (
+	compressMethodGZIP = "gzip"
+)
+
+func Load(raw []byte, d any) (err error) {
+	if len(raw) == 0 {
+		return nil
+	}
+	var payload compressPayload
+	_ = json.Unmarshal(raw, &payload)
+	var r io.Reader
+	switch payload.CompressMethod {
+	case compressMethodGZIP:
+		r, err = gzip.NewReader(bytes.NewReader(payload.Data))
+		if err != nil {
+			return errors.Wrapf(err, "try to load as compressed payload failed")
+		}
+	default:
+		r = bytes.NewReader(raw)
+	}
+	return json.NewDecoder(r).Decode(d)
+}
+
+func Dump(d any) ([]byte, error) {
+	return dump(d, compressMethodGZIP)
+}
+
+func dump(d any, compressMethod string) ([]byte, error) {
+	// prepare WriteCloser by compressMethod
+	var buf bytes.Buffer
+	var w io.WriteCloser
+	switch compressMethod {
+	case compressMethodGZIP:
+		w = gzip.NewWriter(&buf)
+	default:
+		return nil, errors.Errorf("compress method %s not supported", compressMethod)
+	}
+	// json marshal and do compress
+	err := json.NewEncoder(w).Encode(d)
+	if err == nil {
+		err = w.Close()
+	}
+	if err != nil {
+		return nil, errors.Wrapf(err, "dump with compress method %s failed", compressMethod)
+	}
+	// build payload and json marshal
+	var raw []byte
+	raw, err = json.Marshal(compressPayload{
+		CompressMethod: compressMethod,
+		Data:           buf.Bytes(),
+	})
+	if err != nil {
+		err = errors.Wrapf(err, "dump with compress method %s failed", compressMethod)
+	}
+	return raw, err
+}

common/compress/compress_test.go

@@ -0,0 +1,32 @@
+package compress
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func Test_loadAndDump(t *testing.T) {
+	const count = 100000
+	origin := make([]string, count)
+	for i := 0; i < count; i++ {
+		origin[i] = fmt.Sprintf("%08d", i)
+	}
+
+	d0, _ := json.Marshal(origin)
+	t.Logf("origin len: %d", len(d0))
+	d, err := Dump(origin)
+	assert.NoError(t, err)
+
+	t.Logf("dump len: %d", len(d))
+	t.Logf("dump prefix: %s", string(d[:100]))
+	t.Logf("dump suffix: %s", string(d[len(d)-100:]))
+
+	var result []string
+	assert.NoError(t, Load(d, &result))
+	assert.Equal(t, origin, result)
+
+	assert.NoError(t, Load(d0, &result))
+	assert.Equal(t, origin, result)
+}

common/contract/BUILD.bazel

@@ -0,0 +1,19 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "contract",
+    srcs = ["contract.go"],
+    importpath = "sentioxyz/sentio-core/common/contract",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//common/https",
+        "@com_github_pkg_errors//:errors",
+    ],
+)
+
+go_test(
+    name = "contract_test",
+    srcs = ["contract_test.go"],
+    embed = [":contract"],
+    deps = ["@com_github_stretchr_testify//assert"],
+)

common/contract/contract.go

@@ -0,0 +1,148 @@
+package contract
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+
+	"github.com/pkg/errors"
+
+	"sentioxyz/sentio-core/common/https"
+)
+
+type TokenResult struct {
+	Decimals *int    `json:"decimals"`
+	Logo     *string `json:"logo"`
+	Name     string  `json:"name"`
+	Symbol   string  `json:"symbol"`
+}
+
+type jsonError struct {
+	Code    int         `json:"code"`
+	Message string      `json:"message"`
+	Data    interface{} `json:"data,omitempty"`
+}
+
+type JsonrpcMessage struct {
+	Version string          `json:"jsonrpc,omitempty"`
+	ID      json.RawMessage `json:"id,omitempty"`
+	Method  string          `json:"method,omitempty"`
+	Params  json.RawMessage `json:"params,omitempty"`
+	Error   *jsonError      `json:"error,omitempty"`
+	Result  *TokenResult    `json:"result,omitempty"`
+}
+
+// IsERC20 refers from https://docs.alchemy.com/reference/alchemy-gettokenmetadata
+func IsERC20(ctx context.Context, apiEndpoint string, tokenAddress string) (bool, error) {
+	payload := strings.NewReader(
+		fmt.Sprintf(
+			"{\"id\":1,\"jsonrpc\":\"2.0\",\"method\":\"alchemy_getTokenMetadata\",\"params\":[\"%s\"]}",
+			tokenAddress,
+		),
+	)
+	req, _ := http.NewRequestWithContext(ctx, http.MethodPost, apiEndpoint, payload)
+
+	req.Header.Add("accept", "application/json")
+	req.Header.Add("content-type", "application/json")
+
+	res, err := https.DefaultClient.Do(req)
+
+	if err != nil {
+		return false, errors.WithStack(err)
+	}
+	defer res.Body.Close()
+
+	if res.StatusCode != http.StatusOK {
+		return false, errors.New("HTTP request return" + res.Status)
+	}
+
+	body, _ := io.ReadAll(res.Body)
+	//bodyString := string(body)
+
+	var jsonrpcRes JsonrpcMessage
+	err = json.Unmarshal(body, &jsonrpcRes)
+	if err != nil {
+		return false, err
+	}
+
+	if jsonrpcRes.Result != nil {
+		if jsonrpcRes.Result.Decimals != nil {
+			return true, nil
+		}
+	}
+
+	return false, nil
+}
+
+// IsERC20New refers from https://docs.moralis.io/reference/gettokenmetadata
+func IsERC20New(ctx context.Context, apiKey string, chainID string, tokenAddress string) (bool, error) {
+	info, err := getERC20Info(ctx, apiKey, chainID, []string{tokenAddress})
+	if err != nil {
+		return false, err
+	}
+	return info[0].Symbol != "" && info[0].Decimals != "", nil
+}
+
+type ERC20Token struct {
+	Address     string `json:"address"`
+	Symbol      string `json:"symbol"`
+	Name        string `json:"name"`
+	Decimals    string `json:"decimals"`
+	Logo        string `json:"logo,omitempty"`
+	LogoHash    string `json:"logo_hash,omitempty"`
+	Thumbnail   string `json:"thumbnail,omitempty"`
+	BlockNumber string `json:"block_number,omitempty"`
+	//Validated   string `json:"validated,omitempty"`
+}
+
+// https://docs.moralis.io/reference/gettokenmetadata
+func getERC20Info(ctx context.Context, apiKey string, chainID string, tokenAddress []string) ([]ERC20Token, error) {
+	var chain string
+	switch chainID {
+	case "1":
+		chain = "eth"
+	case "5":
+		chain = "goerli"
+	case "56":
+		chain = "bsc"
+	// TDDO add more mappings
+	default:
+		return nil, errors.New("chainID not supported")
+	}
+
+	var addresses = ""
+	for _, address := range tokenAddress {
+		addresses += "&addresses=" + address
+	}
+
+	url := "https://deep-index.moralis.io/api/v2/erc20/metadata?&chain=" + chain + addresses
+
+	req, _ := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+
+	req.Header.Add("accept", "application/json")
+	req.Header.Add("X-API-Key", apiKey)
+
+	res, err := https.DefaultClient.Do(req)
+
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	defer res.Body.Close()
+
+	if res.StatusCode != http.StatusOK {
+		return nil, errors.New("HTTP request return" + res.Status)
+	}
+
+	body, _ := io.ReadAll(res.Body)
+
+	var tokenInfo []ERC20Token
+	err = json.Unmarshal(body, &tokenInfo)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return tokenInfo, nil
+}

common/contract/contract_test.go

@@ -0,0 +1,83 @@
+package contract
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// FIXME runs into 403 regularly
+//func TestIsERC20(t *testing.T) {
+//	endpoint := "https://eth-mainnet.g.alchemy.com/v2/KLPDGUUQGKmScCbSdPd0iUNO_JufXdg9"
+//
+//	ctx := context.Background()
+//
+//	var res bool
+//	var err error
+//
+//	//// abtc
+//	//res, err = IsERC20(ctx, endpoint, "0xC2fcab14Ec1F2dFA82a23C639c4770345085a50F")
+//	//assert.NoError(t, err)
+//	//assert.False(t, res)
+//
+//	// weth address
+//	res, err = IsERC20(ctx, endpoint, "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2")
+//	assert.NoError(t, err)
+//	assert.True(t, res)
+//
+//	// invalid address
+//	res, err = IsERC20(ctx, endpoint, "0x0000000000000000000000000000000000000000")
+//	assert.NoError(t, err)
+//	assert.False(t, res)
+//
+//	// token address but does not implement ERC20
+//	res, err = IsERC20(ctx, endpoint, "0xa6794DEc66Df7d8B69752956df1b28cA93f77CD7")
+//	assert.NoError(t, err)
+//	assert.False(t, res)
+//
+//	// USDC
+//	res, err = IsERC20(ctx, endpoint, "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48")
+//	assert.NoError(t, err)
+//	assert.True(t, res)
+//
+//	res, err = IsERC20(ctx, endpoint, "0x6B89B97169a797d94F057F4a0B01E2cA303155e4")
+//	assert.NoError(t, err)
+//	assert.True(t, res)
+//}
+
+func TestIsERC20New(t *testing.T) {
+	//endpoint := "https://eth-mainnet.g.alchemy.com/v2/z1Q-YhcYg60C5sOQPUzsMFqiDJSvqbsK"
+	endpoint := "oGDQZsjYX3IdnfkenzRf0K5NA7Lsy0NljUFVKFp0nGDhwajEq5ltjHU7aFp3V8lG"
+
+	ctx := context.Background()
+
+	var res bool
+	var err error
+
+	//// abtc
+	//res, err = IsERC20New(ctx, endpoint, "1", "0xC2fcab14Ec1F2dFA82a23C639c4770345085a50F")
+	//assert.NoError(t, err)
+	//assert.False(t, res)
+
+	// weth address
+	res, err = IsERC20New(ctx, endpoint, "1", "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2")
+	assert.NoError(t, err)
+	assert.True(t, res)
+
+	// invalid address
+	// res, err = IsERC20New(ctx, endpoint, "1", "0x0000000000000000000000000000000000000000")
+	// assert.NoError(t, err)
+	// assert.False(t, res)
+
+	// token address but does not implement ERC20
+	_, err = IsERC20New(ctx, endpoint, "1", "0xa6794DEc66Df7d8B69752956df1b28cA93f77CD7")
+	assert.NoError(t, err)
+	// Uncomment this line when the issue is fixed.
+	// assert.False(t, res)
+
+	// USDC
+	res, err = IsERC20New(ctx, endpoint, "1", "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48")
+	assert.NoError(t, err)
+	assert.True(t, res)
+}