From 1354108fc9af5326f7be5091dd05d5affabfb65c Mon Sep 17 00:00:00 2001 From: Ze Date: Wed, 21 Aug 2024 16:06:36 -0700 Subject: [PATCH 1/2] refactor(lib): remove unused folders --- lib/evmchain/evmchain.go | 37 +- lib/fireblocks/README.md | 14 - lib/fireblocks/account.go | 149 -------- lib/fireblocks/client.go | 138 -------- lib/fireblocks/client_internal_test.go | 28 -- lib/fireblocks/client_test.go | 162 --------- lib/fireblocks/config.go | 93 ----- lib/fireblocks/gettransaction.go | 37 -- lib/fireblocks/jsonhttp.go | 104 ------ lib/fireblocks/jwt.go | 43 --- lib/fireblocks/key.go | 32 -- lib/fireblocks/sign.go | 153 --------- lib/fireblocks/supportedassets.go | 35 -- lib/fireblocks/testdata/test_private_key.pem | 52 --- lib/fireblocks/types.go | 303 ---------------- lib/forkjoin/forkjoin.go | 264 -------------- lib/forkjoin/forkjoin_test.go | 142 -------- lib/merkle/README.md | 5 - lib/merkle/core.go | 198 ----------- lib/merkle/core_internal_test.go | 102 ------ lib/merkle/core_test.go | 87 ----- lib/stream/stream.go | 341 ------------------- lib/tokens/coingecko/coingecko.go | 107 ------ lib/tokens/coingecko/coingecko_test.go | 65 ---- lib/tokens/coingecko/options.go | 17 - lib/tokens/mock.go | 41 --- lib/tokens/price.go | 57 ---- lib/tokens/tokens.go | 42 --- 28 files changed, 20 insertions(+), 2828 deletions(-) delete mode 100644 lib/fireblocks/README.md delete mode 100644 lib/fireblocks/account.go delete mode 100644 lib/fireblocks/client.go delete mode 100644 lib/fireblocks/client_internal_test.go delete mode 100644 lib/fireblocks/client_test.go delete mode 100644 lib/fireblocks/config.go delete mode 100644 lib/fireblocks/gettransaction.go delete mode 100644 lib/fireblocks/jsonhttp.go delete mode 100644 lib/fireblocks/jwt.go delete mode 100644 lib/fireblocks/key.go delete mode 100644 lib/fireblocks/sign.go delete mode 100644 lib/fireblocks/supportedassets.go delete mode 100644 lib/fireblocks/testdata/test_private_key.pem delete mode 100644 lib/fireblocks/types.go delete mode 100644 lib/forkjoin/forkjoin.go delete mode 100644 lib/forkjoin/forkjoin_test.go delete mode 100644 lib/merkle/README.md delete mode 100644 lib/merkle/core.go delete mode 100644 lib/merkle/core_internal_test.go delete mode 100644 lib/merkle/core_test.go delete mode 100644 lib/stream/stream.go delete mode 100644 lib/tokens/coingecko/coingecko.go delete mode 100644 lib/tokens/coingecko/coingecko_test.go delete mode 100644 lib/tokens/coingecko/options.go delete mode 100644 lib/tokens/mock.go delete mode 100644 lib/tokens/price.go delete mode 100644 lib/tokens/tokens.go diff --git a/lib/evmchain/evmchain.go b/lib/evmchain/evmchain.go index 3ee74931..217534f8 100644 --- a/lib/evmchain/evmchain.go +++ b/lib/evmchain/evmchain.go @@ -3,10 +3,10 @@ package evmchain import ( "time" - - "github.com/piplabs/story/lib/tokens" ) +type Token string + const ( // Mainnets. IDEthereum uint64 = 1 @@ -30,15 +30,18 @@ const ( IDMockOp uint64 = 1655 IDMockArb uint64 = 1656 - iliadEVMName = "iliad_evm" + iliadEVMName = "story_evm" iliadEVMBlockPeriod = time.Second * 2 + + IP Token = "IP" + ETH Token = "ETH" ) type Metadata struct { ChainID uint64 Name string BlockPeriod time.Duration - NativeToken tokens.Token + NativeToken Token } func MetadataByID(chainID uint64) (Metadata, bool) { @@ -61,78 +64,78 @@ var static = map[uint64]Metadata{ ChainID: IDEthereum, Name: "ethereum", BlockPeriod: 12 * time.Second, - NativeToken: tokens.ETH, + NativeToken: ETH, }, IDIliadMainnet: { ChainID: IDIliadMainnet, Name: iliadEVMName, BlockPeriod: iliadEVMBlockPeriod, - NativeToken: tokens.ILIAD, + NativeToken: IP, }, IDIliadTestnet: { ChainID: IDIliadTestnet, Name: iliadEVMName, BlockPeriod: iliadEVMBlockPeriod, - NativeToken: tokens.ILIAD, + NativeToken: IP, }, IDIliad: { ChainID: IDIliad, Name: iliadEVMName, BlockPeriod: iliadEVMBlockPeriod, - NativeToken: tokens.ILIAD, + NativeToken: IP, }, IDHolesky: { ChainID: IDHolesky, Name: "holesky", BlockPeriod: 12 * time.Second, - NativeToken: tokens.ETH, + NativeToken: ETH, }, IDArbSepolia: { ChainID: IDArbSepolia, Name: "arb_sepolia", BlockPeriod: 300 * time.Millisecond, - NativeToken: tokens.ETH, + NativeToken: ETH, }, IDOpSepolia: { ChainID: IDOpSepolia, Name: "op_sepolia", BlockPeriod: 2 * time.Second, - NativeToken: tokens.ETH, + NativeToken: ETH, }, IDIliadEphemeral: { ChainID: IDIliadEphemeral, Name: iliadEVMName, BlockPeriod: iliadEVMBlockPeriod, - NativeToken: tokens.ILIAD, + NativeToken: IP, }, IDMockL1Fast: { ChainID: IDMockL1Fast, Name: "mock_l1", BlockPeriod: time.Second, - NativeToken: tokens.ETH, + NativeToken: ETH, }, IDMockL1Slow: { ChainID: IDMockL1Slow, Name: "slow_l1", BlockPeriod: time.Second * 12, - NativeToken: tokens.ETH, + NativeToken: ETH, }, IDMockL2: { ChainID: IDMockL2, Name: "mock_l2", BlockPeriod: time.Second, - NativeToken: tokens.ETH, + NativeToken: ETH, }, IDMockOp: { ChainID: IDMockOp, Name: "mock_op", BlockPeriod: time.Second * 2, - NativeToken: tokens.ETH, + NativeToken: ETH, }, IDMockArb: { ChainID: IDMockArb, Name: "mock_arb", BlockPeriod: time.Second / 4, - NativeToken: tokens.ETH, + NativeToken: ETH, }, } diff --git a/lib/fireblocks/README.md b/lib/fireblocks/README.md deleted file mode 100644 index 3d440a8b..00000000 --- a/lib/fireblocks/README.md +++ /dev/null @@ -1,14 +0,0 @@ -# Fireblocks -https://developers.fireblocks.com/reference/api-overview - -## Transactions - -This is going to be our primary API call. We use raw signing for all of our create transaction requests. This is because we cannot guranteee fireblocks is integrated with the chains we are deploying on. - - -## Request Signing: - -https://developers.fireblocks.com/reference/signing-a-request-jwt-structure - -### Note: -A deployment transaction is no different than a normal transaction, we build a transaction payload (the payload just specifies to deploy), we send the txn payload to fireblocks to be signed, we get it back and then send it to the chain. diff --git a/lib/fireblocks/account.go b/lib/fireblocks/account.go deleted file mode 100644 index fe8f466c..00000000 --- a/lib/fireblocks/account.go +++ /dev/null @@ -1,149 +0,0 @@ -package fireblocks - -import ( - "bytes" - "context" - "crypto/ecdsa" - "encoding/hex" - "net/http" - "strconv" - "text/template" - - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/crypto" - - "github.com/piplabs/story/lib/errors" -) - -// Accounts returns all the vault accounts from the account cache, populating it if empty. -func (c Client) Accounts(ctx context.Context) (map[common.Address]uint64, error) { - if err := c.cache.MaybePopulate(ctx, c.queryAccounts); err != nil { - return nil, errors.Wrap(err, "populating account cache") - } - - return c.cache.Clone(), nil -} - -// getAccount returns the Fireblocks account ID for the given address from the account cache. -// It populates the cache if the account is not found. -func (c Client) getAccount(ctx context.Context, addr common.Address) (uint64, error) { - accounts, err := c.Accounts(ctx) - if err != nil { - return 0, err - } - - account, ok := accounts[addr] - if !ok { - return 0, errors.New("account not found") - } - - return account, nil -} - -// queryAccounts returns all the vault accounts from the Fireblocks API. -func (c Client) queryAccounts(ctx context.Context) (map[common.Address]uint64, error) { - header, err := c.authHeaders(endpointVaults, nil) - if err != nil { - return nil, err - } - - var resp vaultsResponse - var errResp errorResponse - ok, err := c.jsonHTTP.Send( - ctx, - endpointVaults, - http.MethodGet, - nil, - header, - &resp, - &errResp, - ) - if err != nil { - return nil, err - } else if !ok { - return nil, errors.New("failed to get vaults", "resp_msg", errResp.Message, "resp_code", errResp.Code) - } else if resp.Paging.After != "" { - return nil, errors.New("paging not implemented") - } - - accounts := make(map[common.Address]uint64, len(resp.Accounts)) - for _, account := range resp.Accounts { - id, err := strconv.ParseUint(account.ID, 10, 64) - if err != nil { - return nil, errors.Wrap(err, "parsing account ID") - } - - pubkey, err := c.GetPublicKey(ctx, id) - if err != nil { - return nil, errors.Wrap(err, "getting public key") - } - - accounts[crypto.PubkeyToAddress(*pubkey)] = id - } - - return accounts, nil -} - -// GetPublicKey returns the public key for the given vault account. -func (c Client) GetPublicKey(ctx context.Context, account uint64) (*ecdsa.PublicKey, error) { - endpoint, err := c.pubkeyEndpoint(account) - if err != nil { - return nil, errors.Wrap(err, "getting pubkey endpoint") - } - - headers, err := c.authHeaders(endpoint, nil) - if err != nil { - return nil, err - } - - var res pubkeyResponse - var errRes errorResponse - ok, err := c.jsonHTTP.Send( - ctx, - endpoint, - http.MethodGet, - nil, - headers, - &res, - &errRes, - ) - if err != nil { - return nil, err - } else if !ok { - return nil, errors.New("failed to get public key", "resp_msg", errRes.Message, "resp_code", errRes.Code) - } - - pk, err := hex.DecodeString(res.PublicKey) - if err != nil { - return nil, errors.Wrap(err, "decoding public key") - } - - resp, err := crypto.DecompressPubkey(pk) - if err != nil { - return nil, errors.Wrap(err, "decompressing public key") - } - - return resp, nil -} - -// pubkeyEndpoint returns the public key endpoint by populating the template. -func (c Client) pubkeyEndpoint(account uint64) (string, error) { - tmpl, err := template.New("").Parse(endpointPubkeyTmpl) - if err != nil { - return "", errors.Wrap(err, "parsing pubkey endpoint template") - } - - var buf bytes.Buffer - err = tmpl.Execute(&buf, struct { - VaultAccountID string - AssetID string - }{ - VaultAccountID: strconv.FormatUint(account, 10), - AssetID: c.getAssetID(), - }) - if err != nil { - return "", errors.Wrap(err, "executing pubkey endpoint template") - } - - return buf.String(), nil -} diff --git a/lib/fireblocks/client.go b/lib/fireblocks/client.go deleted file mode 100644 index 91238cac..00000000 --- a/lib/fireblocks/client.go +++ /dev/null @@ -1,138 +0,0 @@ -package fireblocks - -import ( - "context" - "crypto/rsa" - "maps" - "sync" - - "github.com/ethereum/go-ethereum/common" - - "github.com/piplabs/story/lib/errors" - "github.com/piplabs/story/lib/netconf" -) - -const ( - endpointTransactions = "/v1/transactions" - endpointAssets = "/v1/supported_assets" - endpointVaults = "/v1/vault/accounts_paged" - endpointPubkeyTmpl = "/v1/vault/accounts/{{.VaultAccountID}}/{{.AssetID}}/0/0/public_key_info?compressed" - - assetHolesky = "ETH_TEST6" - assetSepolia = "ETH_TEST5" - assetMainnet = "ETH" - - hostProd = "https://api.fireblocks.io" - hostSandbox = "https://sandbox-api.fireblocks.io" -) - -// Client is a JSON HTTP client for the FireBlocks API. -type Client struct { - opts options - apiKey string - network netconf.ID - privateKey *rsa.PrivateKey - jsonHTTP jsonHTTP - cache *accountCache -} - -// New creates a new FireBlocks client. -func New(network netconf.ID, apiKey string, privateKey *rsa.PrivateKey, opts ...func(*options)) (Client, error) { - if apiKey == "" { - return Client{}, errors.New("apiKey is required") - } - if privateKey == nil { - return Client{}, errors.New("privateKey is required") - } - - o := defaultOptions() - for _, opt := range opts { - opt(&o) - } - if err := o.check(); err != nil { - return Client{}, errors.Wrap(err, "options check") - } - - return Client{ - apiKey: apiKey, - privateKey: privateKey, - jsonHTTP: newJSONHTTP(o.Host, apiKey), - opts: o, - cache: newAccountCache(o.TestAccounts), - network: network, - }, nil -} - -// authHeaders returns the authentication headers for the FireBlocks API. -func (c Client) authHeaders(endpoint string, request any) (map[string]string, error) { - token, err := c.token(endpoint, request) - if err != nil { - return nil, errors.Wrap(err, "generating token") - } - - return map[string]string{ - "X-API-KEY": c.apiKey, - "Authorization": "Bearer " + token, - }, nil -} - -func (c Client) getAssetID() string { - switch c.network { - case netconf.Mainnet: - return assetMainnet - default: - return assetHolesky - } -} - -func newAccountCache(init map[common.Address]uint64) *accountCache { - return &accountCache{ - accountsByAddress: init, - } -} - -type accountCache struct { - sync.Mutex - accountsByAddress map[common.Address]uint64 -} - -func (c *accountCache) MaybePopulate(ctx context.Context, fn func(context.Context) (map[common.Address]uint64, error)) error { - c.Lock() - defer c.Unlock() - - if len(c.accountsByAddress) > 0 { - return nil - } - - accounts, err := fn(ctx) - if err != nil { - return err - } - - c.accountsByAddress = accounts - - return nil -} - -func (c *accountCache) Get(addr common.Address) (uint64, bool) { - c.Lock() - defer c.Unlock() - - acc, ok := c.accountsByAddress[addr] - - return acc, ok -} - -func (c *accountCache) Set(addr common.Address, id uint64) { - c.Lock() - defer c.Unlock() - - c.accountsByAddress[addr] = id -} - -func (c *accountCache) Clone() map[common.Address]uint64 { - c.Lock() - defer c.Unlock() - - return maps.Clone(c.accountsByAddress) -} diff --git a/lib/fireblocks/client_internal_test.go b/lib/fireblocks/client_internal_test.go deleted file mode 100644 index 3e540345..00000000 --- a/lib/fireblocks/client_internal_test.go +++ /dev/null @@ -1,28 +0,0 @@ -package fireblocks - -import ( - "crypto/ecdsa" - "encoding/hex" - "testing" - - "github.com/ethereum/go-ethereum/crypto" -) - -// TransactionResponse returns a transaction response for testing purposes. -func TransactionResponseForT(t *testing.T, id string, sig [65]byte, pubkey *ecdsa.PublicKey) transaction { - t.Helper() - - return transaction{ - ID: id, - Status: "COMPLETED", - SignedMessages: []signedMessage{{ - PublicKey: hex.EncodeToString(crypto.CompressPubkey(pubkey)), - Signature: signature{ - FullSig: hex.EncodeToString(sig[:64]), - R: hex.EncodeToString(sig[:32]), - S: hex.EncodeToString(sig[32:64]), - V: int(sig[64]), - }, - }}, - } -} diff --git a/lib/fireblocks/client_test.go b/lib/fireblocks/client_test.go deleted file mode 100644 index 017dc372..00000000 --- a/lib/fireblocks/client_test.go +++ /dev/null @@ -1,162 +0,0 @@ -package fireblocks_test - -import ( - "context" - "crypto/rsa" - "crypto/x509" - "encoding/json" - "encoding/pem" - "net/http" - "net/http/httptest" - "os" - "strings" - "testing" - "time" - - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/common/hexutil" - "github.com/ethereum/go-ethereum/crypto" - "github.com/google/uuid" - "github.com/stretchr/testify/require" - - "github.com/piplabs/story/lib/fireblocks" - "github.com/piplabs/story/lib/netconf" - "github.com/piplabs/story/lib/tutil" - - _ "embed" -) - -//go:embed testdata/test_private_key.pem -var testPrivateKey []byte - -func TestSignOK(t *testing.T) { - t.Parallel() - - ctx := context.Background() - apiKey := uuid.New().String() - txID := uuid.New().String() - - // Create a private key and sign an expected message - privKey, err := crypto.GenerateKey() - require.NoError(t, err) - addr := crypto.PubkeyToAddress(privKey.PublicKey) - digest := crypto.Keccak256([]byte("test")) - expectSig, err := crypto.Sign(digest, privKey) - require.NoError(t, err) - - // Start a test http server that serves the expected responses - var count int - ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - count++ - if count <= 2 { - // Just return txID and "submitted" on first two attempts - bz, _ := json.Marshal(struct { - ID string `json:"id"` - Status string `json:"status"` - }{ - ID: txID, - Status: "SUBMITTED", - }) - _, _ = w.Write(bz) - - return - } - - // Then return the signed transaction - bz, _ := json.Marshal(fireblocks.TransactionResponseForT(t, txID, [65]byte(expectSig), &privKey.PublicKey)) - _, _ = w.Write(bz) - })) - defer ts.Close() - - client, err := fireblocks.New(netconf.Simnet, apiKey, parseKey(t, testPrivateKey), - fireblocks.WithHost(ts.URL), // Use the test server for all requests. - fireblocks.WithQueryInterval(time.Millisecond), // Fast timeout and interval for testing - fireblocks.WithLogFreqFactor(1), - fireblocks.WithTestAccount(addr, 0), - ) - require.NoError(t, err) - - actualSig, err := client.Sign(ctx, [32]byte(digest), addr) - require.NoError(t, err) - - require.Equal(t, [65]byte(expectSig), actualSig) -} - -func parseKey(t *testing.T, data []byte) *rsa.PrivateKey { - t.Helper() - - p, _ := pem.Decode(data) - k, err := x509.ParsePKCS8PrivateKey(p.Bytes) - require.NoError(t, err) - - return k.(*rsa.PrivateKey) //nolint:forcetypeassert // parseKey is only used for testing -} - -// Populate this or run TestSmoke via terminal -// func init() { -// os.Setenv("FIREBLOCKS_APIKEY", "") -// os.Setenv("FIREBLOCKS_KEY_PATH", "") -//} - -func TestSmoke(t *testing.T) { - t.Parallel() - ctx := context.Background() - - apiKey, ok := os.LookupEnv("FIREBLOCKS_APIKEY") - if !ok { - t.Skip("FIREBLOCKS_APIKEY not set") - } - privKeyFile, ok := os.LookupEnv("FIREBLOCKS_KEY_PATH") - if !ok { - t.Skip("FIREBLOCKS_KEY_PATH not set") - } - privKey, err := os.ReadFile(privKeyFile) - require.NoError(t, err) - - client, err := fireblocks.New(netconf.Staging, apiKey, parseKey(t, privKey)) - require.NoError(t, err) - - addr := common.BytesToAddress(hexutil.MustDecode("0x7a6cF389082dc698285474976d7C75CAdE08ab7e")) - - t.Run("assets", func(t *testing.T) { - t.Parallel() - - assets, err := client.GetSupportedAssets(ctx) - require.NoError(t, err) - - for i, asset := range assets { - if !strings.Contains(asset.ID, "ETH_TEST") { - continue - } - t.Logf("asset %d: %#v", i, asset) - } - }) - - t.Run("accounts", func(t *testing.T) { - t.Parallel() - - accounts, err := client.Accounts(ctx) - tutil.RequireNoError(t, err) - - t.Logf("accounts: %#v", accounts) - }) - - t.Run("pubkey", func(t *testing.T) { - t.Parallel() - - pubkey, err := client.GetPublicKey(ctx, 0) - tutil.RequireNoError(t, err) - - t.Logf("address: %#v", crypto.PubkeyToAddress(*pubkey)) - }) - - t.Run("sign", func(t *testing.T) { - t.Parallel() - - ctx, cancel := context.WithTimeout(ctx, 20*time.Second) - defer cancel() - - _, err = client.Sign(ctx, tutil.RandomHash(), addr) - tutil.RequireNoError(t, err) - }) -} diff --git a/lib/fireblocks/config.go b/lib/fireblocks/config.go deleted file mode 100644 index cb067ddf..00000000 --- a/lib/fireblocks/config.go +++ /dev/null @@ -1,93 +0,0 @@ -package fireblocks - -import ( - "time" - - "github.com/ethereum/go-ethereum/common" - - "github.com/piplabs/story/lib/errors" -) - -// options houses parameters for altering the behavior of a SimpleTxManager. -type options struct { - // NetworkTimeout is the allowed duration for a single network request. - // This is intended to be used for network requests that can be replayed. - NetworkTimeout time.Duration - - // QueryInterval is the interval at which the FireBlocks client will - // call the get transaction by id to check for confirmations after a txn - // has been sent - QueryInterval time.Duration - - // LogFreqFactor is the frequency at which the FireBlocks client will - // log a warning message if the transaction has not been signed yet - LogFreqFactor int - - // SignNote is a note to include in the sign request - SignNote string - - // Host is the base URL for the FireBlocks API. - Host string - - // TestAccounts overrides dynamic account - TestAccounts map[common.Address]uint64 -} - -// defaultOptions returns a options with default values. -func defaultOptions() options { - return options{ - NetworkTimeout: time.Duration(30) * time.Second, - QueryInterval: time.Second, - LogFreqFactor: 10, - Host: hostProd, - TestAccounts: make(map[common.Address]uint64), - SignNote: "iliad sign note not set", - } -} - -func WithQueryInterval(interval time.Duration) func(*options) { - return func(cfg *options) { - cfg.QueryInterval = interval - } -} - -func WithLogFreqFactor(factor int) func(*options) { - return func(cfg *options) { - cfg.LogFreqFactor = factor - } -} - -func WithTestAccount(addr common.Address, accID uint64) func(*options) { - return func(cfg *options) { - cfg.TestAccounts[addr] = accID - } -} - -func WithHost(host string) func(*options) { - return func(cfg *options) { - cfg.Host = host - } -} - -func WithSignNote(note string) func(*options) { - return func(cfg *options) { - cfg.SignNote = note - } -} - -// check validates the options. -func (c options) check() error { - if c.LogFreqFactor <= 0 { - return errors.New("must provide LogFreqFactor") - } - - if c.NetworkTimeout <= 0 { - return errors.New("must provide NetworkTimeout") - } - - if c.QueryInterval <= 0 { - return errors.New("must provide QueryInterval") - } - - return nil -} diff --git a/lib/fireblocks/gettransaction.go b/lib/fireblocks/gettransaction.go deleted file mode 100644 index 2d662e20..00000000 --- a/lib/fireblocks/gettransaction.go +++ /dev/null @@ -1,37 +0,0 @@ -package fireblocks - -import ( - "context" - "net/http" - "path/filepath" - - "github.com/piplabs/story/lib/errors" -) - -// getTransactionByID gets a transaction by its ID. -func (c Client) getTransactionByID(ctx context.Context, transactionID string) (transaction, error) { - endpoint := filepath.Join(endpointTransactions, transactionID) - headers, err := c.authHeaders(endpoint, nil) - if err != nil { - return transaction{}, err - } - - var res transaction - var errRes errorResponse - ok, err := c.jsonHTTP.Send( - ctx, - endpoint, - http.MethodGet, - nil, - headers, - &res, - &errRes, - ) - if err != nil { - return transaction{}, err - } else if !ok { - return transaction{}, errors.New("failed to get transaction", "resp_msg", errRes.Message, "resp_code", errRes.Code) - } - - return res, nil -} diff --git a/lib/fireblocks/jsonhttp.go b/lib/fireblocks/jsonhttp.go deleted file mode 100644 index 63e38898..00000000 --- a/lib/fireblocks/jsonhttp.go +++ /dev/null @@ -1,104 +0,0 @@ -package fireblocks - -import ( - "bytes" - "context" - "encoding/json" - "io" - "net/http" - "net/url" - - "github.com/piplabs/story/lib/errors" -) - -// jsonHTTP provides a simple interface for sending JSON HTTP requests. -type jsonHTTP struct { - apiKey string - host string - http http.Client -} - -// newJSONHTTP creates a new jsonHTTP. -func newJSONHTTP(host string, apiKey string) jsonHTTP { - return jsonHTTP{ - host: host, - apiKey: apiKey, - } -} - -// Send sends an JSON HTTP request with the json formatted request as body. -// If the response status code is 2XX, it marshals the response body into the response pointer and returns true. -// Else, it marshals the response body into the errResponse pointer and returns false. -func (c jsonHTTP) Send(ctx context.Context, uri string, httpMethod string, request any, headers map[string]string, response any, errResponse any) (bool, error) { - endpoint, err := url.Parse(c.host + uri) - if err != nil { - return false, errors.Wrap(err, "parse") - } - - // on get requests even will a nil request, we are passing in a non nil request body as the body marshaled to equal `null` - // so we just set it to nil if the request is nil - var reqBytes []byte - if request != nil { - reqBytes, err = json.Marshal(request) - if err != nil { - return false, errors.Wrap(err, "marshaling JSON") - } - } - - req, err := http.NewRequestWithContext( - ctx, - httpMethod, - endpoint.String(), - bytes.NewReader(reqBytes), - ) - if err != nil { - return false, errors.Wrap(err, "new http request") - } - - req.Header = mergeJSONHeaders(headers) - - resp, err := c.http.Do(req) - if err != nil { - return false, errors.Wrap(err, "http do") - } - defer resp.Body.Close() - - respBytes, err := io.ReadAll(resp.Body) - if err != nil { - return false, errors.Wrap(err, "read response body") - } - - if resp.StatusCode/http.StatusContinue != 2 { - if errResponse != nil { - err = json.Unmarshal(respBytes, errResponse) - if err != nil { - return false, errors.Wrap(err, "unmarshal error response", "status code", resp.StatusCode, "body", string(respBytes)) - } - } - - return false, nil - } - - if response != nil { - err = json.Unmarshal(respBytes, response) - if err != nil { - return false, errors.Wrap(err, "unmarshal response") - } - } - - return true, nil -} - -// mergeJSONHeaders merges the default JSON headers with the given headers. -func mergeJSONHeaders(m map[string]string) http.Header { - header := http.Header{} - - header.Set("Accept", "application/json") - header.Set("Content-Type", "application/json") - - for k, v := range m { - header.Set(k, v) - } - - return header -} diff --git a/lib/fireblocks/jwt.go b/lib/fireblocks/jwt.go deleted file mode 100644 index f55fad2d..00000000 --- a/lib/fireblocks/jwt.go +++ /dev/null @@ -1,43 +0,0 @@ -package fireblocks - -import ( - "crypto/sha256" - "encoding/hex" - "encoding/json" - "time" - - "github.com/golang-jwt/jwt/v5" - "github.com/google/uuid" - - "github.com/piplabs/story/lib/errors" -) - -const validFor = 29 * time.Second // Must be less than 30sec. - -// token generates a JWT token for the Fireblocks API. -func (c Client) token(uri string, reqBody any) (string, error) { - nonce := uuid.New().String() - - bz, err := json.Marshal(reqBody) - if err != nil { - return "", errors.Wrap(err, "marshaling JSON") - } - reqHash := sha256.Sum256(bz) - - claims := jwt.MapClaims{ - "uri": uri, // uri - The URI part of the request (e.g., /v1/transactions?foo=bar). - "nonce": nonce, // nonce - Unique number or string. Each API request needs to have a different nonce. - "iat": time.Now().Unix(), // iat - The time at which the JWT was issued, in seconds since Epoch. - "exp": time.Now().Add(validFor).Unix(), // exp - The expiration time on and after which the JWT must not be accepted for processing, in seconds since Epoch. (Must be less than iat+30sec.) - "sub": c.apiKey, // sub - The API Key. - "bodyHash": hex.EncodeToString(reqHash[:]), // bodyHash - Hex-encoded SHA-256 hash of the raw HTTP request body. - } - - token := jwt.NewWithClaims(jwt.SigningMethodRS256, claims) - tokenString, err := token.SignedString(c.privateKey) - if err != nil { - return "", errors.Wrap(err, "signing token") - } - - return tokenString, nil -} diff --git a/lib/fireblocks/key.go b/lib/fireblocks/key.go deleted file mode 100644 index 5f7f57c2..00000000 --- a/lib/fireblocks/key.go +++ /dev/null @@ -1,32 +0,0 @@ -package fireblocks - -import ( - "crypto/rsa" - "crypto/x509" - "encoding/pem" - "fmt" - "os" - - "github.com/piplabs/story/lib/errors" -) - -// LoadKey loads and returns the RSA256 from disk. -func LoadKey(path string) (*rsa.PrivateKey, error) { - bz, err := os.ReadFile(path) - if err != nil { - return nil, errors.Wrap(err, "load fireblocks key", "path", path) - } - - p, _ := pem.Decode(bz) - k, err := x509.ParsePKCS8PrivateKey(p.Bytes) - if err != nil { - return nil, errors.Wrap(err, "parse fireblocks key") - } - - resp, ok := k.(*rsa.PrivateKey) - if !ok { - return nil, errors.New("invalid fireblocks key type", "type", fmt.Sprintf("%T", resp)) - } - - return resp, nil -} diff --git a/lib/fireblocks/sign.go b/lib/fireblocks/sign.go deleted file mode 100644 index 23af4afa..00000000 --- a/lib/fireblocks/sign.go +++ /dev/null @@ -1,153 +0,0 @@ -package fireblocks - -import ( - "context" - "encoding/hex" - "net/http" - "strconv" - "time" - - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/crypto" - - "github.com/piplabs/story/lib/errors" - "github.com/piplabs/story/lib/log" -) - -func (c Client) createRawSignTransaction(ctx context.Context, account uint64, digest common.Hash) (string, error) { - request := c.newRawSignRequest(account, digest) - headers, err := c.authHeaders(endpointTransactions, request) - if err != nil { - return "", err - } - - var res createTransactionResponse - var errRes errorResponse - ok, err := c.jsonHTTP.Send( - ctx, - endpointTransactions, - http.MethodPost, - request, - headers, - &res, - &errRes, - ) - if err != nil { - return "", err - } else if !ok { - return "", errors.New("failed to create transaction", "resp_msg", errRes.Message, "resp_code", errRes.Code) - } - - return res.ID, nil -} - -// Sign creates a raw sign transaction and waits for it to complete, returning the resulting signature (Ethereum RSV format). -// The signer address is checked against the resulting signed address. -func (c Client) Sign(ctx context.Context, digest common.Hash, signer common.Address) ([65]byte, error) { - account, err := c.getAccount(ctx, signer) - if err != nil { - return [65]byte{}, err - } - - id, err := c.createRawSignTransaction(ctx, account, digest) - if err != nil { - return [65]byte{}, errors.Wrap(err, "create raw sign tx") - } - - // First try immediately. - resp, status, err := c.maybeGetSignature(ctx, id, digest, signer) - if err != nil { - return [65]byte{}, errors.Wrap(err, "get sig") - } else if status.Completed() { - return resp, nil - } - - // Then poll every QueryInterval - queryTicker := time.NewTicker(c.opts.QueryInterval) - defer queryTicker.Stop() - - var attempt int - prevStatus := status - for { - select { - case <-ctx.Done(): - return [65]byte{}, errors.Wrap(ctx.Err(), "timeout waiting", "prev_status", prevStatus) - case <-queryTicker.C: - resp, status, err = c.maybeGetSignature(ctx, id, digest, signer) - if err != nil { - return [65]byte{}, errors.Wrap(err, "get sig", "prev_status", prevStatus) - } else if status.Completed() { - return resp, nil - } - - prevStatus = status - - attempt++ - if attempt%c.opts.LogFreqFactor == 0 { - log.Warn(ctx, "Fireblocks transaction not signed yet (will retry)", nil, - "attempt", attempt, - "status", status, - "id", id, - ) - } - } - } -} - -// maybeGetSignature returns the resulting signature and "completed" status if the transaction has been signed. -// If the transaction is still pending, it returns an empty signature and the "pending" status. -// If the transaction has failed, it returns an empty signature and the "failed" status and an error. -func (c Client) maybeGetSignature(ctx context.Context, txID string, digest common.Hash, signer common.Address) ([65]byte, Status, error) { - tx, err := c.getTransactionByID(ctx, txID) - if err != nil { - return [65]byte{}, "", errors.Wrap(err, "get tx") - } - - if tx.Status.Failed() { - return [65]byte{}, tx.Status, errors.New("transaction failed", "status", tx.Status) - } else if !tx.Status.Completed() { - return [65]byte{}, tx.Status, nil - } - // Get the resulting signature. - sig, err := tx.Sig0() - if err != nil { - return [65]byte{}, "", errors.Wrap(err, "get signature") - } - - // Get the signer pubkey. - pubkey, err := tx.Pubkey0() - if err != nil { - return [65]byte{}, "", err - } - addr := crypto.PubkeyToAddress(*pubkey) - if addr != signer { // Ensure it matches the expected signer. - return [65]byte{}, "", errors.New("signed address mismatch", "expect", signer, "actual", addr) - } - - // Ensure the signature is valid. - if !crypto.VerifySignature(crypto.CompressPubkey(pubkey), digest[:], sig[:64]) { - return [65]byte{}, "", errors.New("signature verification failed") - } - - return sig, tx.Status, nil -} - -// newRawSignRequest creates a new transaction request. -func (c Client) newRawSignRequest(account uint64, digest common.Hash) createTransactionRequest { - return createTransactionRequest{ - Operation: "RAW", - Source: source{ - Type: "VAULT_ACCOUNT", - ID: strconv.FormatUint(account, 10), - }, - AssetID: c.getAssetID(), - ExtraParameters: &extraParameters{ - RawMessageData: rawMessageData{ - Messages: []unsignedRawMessage{{ - Content: hex.EncodeToString(digest[:]), // No 0x prefix, just hex. - }}, - }, - }, - Note: c.opts.SignNote, - } -} diff --git a/lib/fireblocks/supportedassets.go b/lib/fireblocks/supportedassets.go deleted file mode 100644 index 97add776..00000000 --- a/lib/fireblocks/supportedassets.go +++ /dev/null @@ -1,35 +0,0 @@ -package fireblocks - -import ( - "context" - "net/http" - - "github.com/piplabs/story/lib/errors" -) - -// GetSupportedAssets returns all asset types supported by Fireblocks. -func (c Client) GetSupportedAssets(ctx context.Context) ([]Asset, error) { - headers, err := c.authHeaders(endpointAssets, nil) - if err != nil { - return nil, err - } - - var res []Asset - var errRes errorResponse - ok, err := c.jsonHTTP.Send( - ctx, - endpointAssets, - http.MethodGet, - nil, - headers, - &res, - &errRes, - ) - if err != nil { - return nil, err - } else if !ok { - return nil, errors.New("failed to get supported assets", "resp_msg", errRes.Message, "resp_code", errRes.Code) - } - - return res, nil -} diff --git a/lib/fireblocks/testdata/test_private_key.pem b/lib/fireblocks/testdata/test_private_key.pem deleted file mode 100644 index ce399555..00000000 --- a/lib/fireblocks/testdata/test_private_key.pem +++ /dev/null @@ -1,52 +0,0 @@ ------BEGIN TESTING KEY----- -MIIJQQIBADANBgkqhkiG9w0BAQEFAASCCSswggknAgEAAoICAQCmH55T2e8fdUaL -iWVL2yI7d/wOu/sxI4nVGoiRMiSMlMZlOEZ4oJY6l2y9N/b8ftwoIpjYO8CBk5au -x2Odgpuz+FJyHppvKakUIeAn4940zoNkRe/iptybIuH5tCBygjs0y1617TlR/c5+ -FF5YRkzsEJrGcLqXzj0hDyrwdplBOv1xz2oHYlvKWWcVMR/qgwoRuj65Ef262t/Q -ELH3+fFLzIIstFTk2co2WaALquOsOB6xGOJSAAr8cIAWe+3MqWM8DOcgBuhABA42 -9IhbBBw0uqTXUv/TGi6tcF29H2buSxAx/Wm6h2PstLd6IJAbWHAa6oTz87H0S6XZ -v42cYoFhHma1OJw4id1oOZMFDTPDbHxgUnr2puSU+Fpxrj9+FWwViKE4j0YatbG9 -cNVpx9xo4NdvOkejWUrqziRorMZTk/zWKz0AkGQzTN3PrX0yy61BoWfznH/NXZ+o -j3PqVtkUs6schoIYvrUcdhTCrlLwGSHhU1VKNGAUlLbNrIYTQNgt2gqvjLEsn4/i -PgS1IsuDHIc7nGjzvKcuR0UeYCDkmBQqKrdhGbdJ1BRohzLdm+woRpjrqmUCbMa5 -VWWldJen0YyAlxNILvXMD117azeduseM1sZeGA9L8MmE12auzNbKr371xzgANSXn -jRuyrblAZKc10kYStrcEmJdfNlzYAwIDAQABAoICABdQBpsD0W/buFuqm2GKzgIE -c4Xp0XVy5EvYnmOp4sEru6/GtvUErDBqwaLIMMv8TY8AU+y8beaBPLsoVg1rn8gg -yAklzExfT0/49QkEDFHizUOMIP7wpbLLsWSmZ4tKRV7CT3c+ZDXiZVECML84lmDm -b6H7feQB2EhEZaU7L4Sc76ZCEkIZBoKeCz5JF46EdyxHs7erE61eO9xqC1+eXsNh -Xr9BS0yWV69K4o/gmnS3p2747AHP6brFWuRM3fFDsB5kPScccQlSyF/j7yK+r+qi -arGg/y+z0+sZAr6gooQ8Wnh5dJXtnBNCxSDJYw/DWHAeiyvk/gsndo3ZONlCZZ9u -bpwBYx3hA2wTa5GUQxFM0KlI7Ftr9Cescf2jN6Ia48C6FcQsepMzD3jaMkLir8Jk -/YD/s5KPzNvwPAyLnf7x574JeWuuxTIPx6b/fHVtboDK6j6XQnzrN2Hy3ngvlEFo -zuGYVvtrz5pJXWGVSjZWG1kc9iXCdHKpmFdPj7XhU0gugTzQ/e5uRIqdOqfNLI37 -fppSuWkWd5uaAg0Zuhd+2L4LG2GhVdfFa1UeHBe/ncFKz1km9Bmjvt04TpxlRnVG -wHxJZKlxpxCZ3AuLNUMP/QazPXO8OIfGOCbwkgFiqRY32mKDUvmEADBBoYpk/wBv -qV99g5gvYFC5Le4QLzOJAoIBAQDcnqnK2tgkISJhsLs2Oj8vEcT7dU9vVnPSxTcC -M0F+8ITukn33K0biUlA+ktcQaF+eeLjfbjkn/H0f2Ajn++ldT56MgAFutZkYvwxJ -2A6PVB3jesauSpe8aqoKMDIj8HSA3+AwH+yU+yA9r5EdUq1S6PscP+5Wj22+thAa -l65CFD77C0RX0lly5zdjQo3Vyca2HYGm/cshFCPRZc66TPjNAHFthbqktKjMQ91H -Hg+Gun2zv8KqeSzMDeHnef4rVaWMIyIBzpu3QdkKPUXMQQxvJ+RW7+MORV9VjE7Z -KVnHa/6x9n+jvtQ0ydHc2n0NOp6BQghTCB2G3w3JJfmPcRSNAoIBAQDAw6mPddoz -UUzANMOYcFtos4EaWfTQE2okSLVAmLY2gtAK6ldTv6X9xl0IiC/DmWqiNZJ/WmVI -glkp6iZhxBSmqov0X9P0M+jdz7CRnbZDFhQWPxSPicurYuPKs52IC08HgIrwErzT -/lh+qRXEqzT8rTdftywj5fE89w52NPHBsMS07VhFsJtU4aY2Yl8y1PHeumXU6h66 -yTvoCLLxJPiLIg9PgvbMF+RiYyomIg75gwfx4zWvIvWdXifQBC88fE7lP2u5gtWL -JUJaMy6LNKHn8YezvwQp0dRecvvoqzoApOuHfsPASHb9cfvcy/BxDXFMJO4QWCi1 -6WLaR835nKLPAoIBAFw7IHSjxNRl3b/FaJ6k/yEoZpdRVaIQHF+y/uo2j10IJCqw -p2SbfQjErLNcI/jCCadwhKkzpUVoMs8LO73v/IF79aZ7JR4pYRWNWQ/N+VhGLDCb -dVAL8x9b4DZeK7gGoE34SfsUfY1S5wmiyiHeHIOazs/ikjsxvwmJh3X2j20klafR -8AJe9/InY2plunHz5tTfxQIQ+8iaaNbzntcXsrPRSZol2/9bX231uR4wHQGQGVj6 -A+HMwsOT0is5Pt7S8WCCl4b13vdf2eKD9xgK4a3emYEWzG985PwYqiXzOYs7RMEV -cgr8ji57aPbRiJHtPbJ/7ob3z5BA07yR2aDz/0kCggEAZDyajHYNLAhHr98AIuGy -NsS5CpnietzNoeaJEfkXL0tgoXxwQqVyzH7827XtmHnLgGP5NO4tosHdWbVflhEf -Z/dhZYb7MY5YthcMyvvGziXJ9jOBHo7Z8Nowd7Rk41x2EQGfve0QcfBd1idYoXch -y47LL6OReW1Vv4z84Szw1fZ0o1yUPVDzxPS9uKP4uvcOevJUh53isuB3nVYArvK5 -p6fjbEY+zaxS33KPdVrajJa9Z+Ptg4/bRqSycTHr2jkN0ZnkC4hkQMH0OfFJb6vD -0VfAaBCZOqHZG/AQ3FFFjRY1P7UEV5WXAn3mKU+HTVJfKug9PxSIvueIttcF3Zm8 -8wKCAQAM43+DnGW1w34jpsTAeOXC5mhIz7J8spU6Uq5bJIheEE2AbX1z+eRVErZX -1WsRNPsNrQfdt/b5IKboBbSYKoGxxRMngJI1eJqyj4LxZrACccS3euAlcU1q+3oN -T10qfQol54KjGld/HVDhzbsZJxzLDqvPlroWgwLdOLDMXhwJYfTnqMEQkaG4Aawr -3P14+Zp/woLiPWw3iZFcL/bt23IOa9YI0NoLhp5MFNXfIuzx2FhVz6BUSeVfQ6Ko -Nx2YZ03g6Kt6B6c43LJx1a/zEPYSZcPERgWOSHlcjmwRfTs6uoN9xt1qs4zEUaKv -Axreud3rJ0rekUp6rI1joG717Wls ------END TESTING KEY----- \ No newline at end of file diff --git a/lib/fireblocks/types.go b/lib/fireblocks/types.go deleted file mode 100644 index e50c36fb..00000000 --- a/lib/fireblocks/types.go +++ /dev/null @@ -1,303 +0,0 @@ -package fireblocks - -import ( - "crypto/ecdsa" - "encoding/hex" - - "github.com/ethereum/go-ethereum/crypto" - - "github.com/piplabs/story/lib/errors" -) - -// Status of a transaction. See https://developers.fireblocks.com/reference/primary-transaction-statuses. -type Status string - -const ( - StatusCompleted Status = "COMPLETED" - StatusFailed Status = "FAILED" - StatusRejected Status = "REJECTED" - StatusBlocked Status = "BLOCKED" - StatusCancelled Status = "CANCELED" - StatusCancelling Status = "CANCELING" - StatusConfirming Status = "CONFIRMING" - StatusBroadcasting Status = "BROADCASTING" - StatusPending3rdParty Status = "PENDING_3RD_PARTY" - StatusPendingSignature Status = "PENDING_SIGNATURE" - StatusQueued Status = "QUEUED" - StatusPendingAuthorization Status = "PENDING_AUTHORIZATION" - StatusPendingAmlScreening Status = "PENDING_AML_SCREENING" - StatusSubmitted Status = "SUBMITTED" -) - -func (s Status) Completed() bool { - return s == StatusCompleted -} - -func (s Status) Failed() bool { - return map[Status]bool{ - StatusFailed: true, - StatusRejected: true, - StatusBlocked: true, - StatusCancelled: true, - StatusCancelling: true, - }[s] -} - -type createTransactionRequest struct { - Operation string `json:"operation"` - Note string `json:"note,omitempty"` - ExternalTxID string `json:"externalTxId,omitempty"` - AssetID string `json:"assetId,omitempty"` - Source source `json:"source"` - Destination *destination `json:"destination,omitempty"` - Destinations []any `json:"destinations,omitempty"` - CustomerRefID string `json:"customerRefId,omitempty"` - Amount string `json:"amountAll,omitempty"` - TreatAsGrossAmount bool `json:"treatAsGrossAmount,omitempty"` - ForceSweep bool `json:"forceSweep,omitempty"` - FeeLevel string `json:"feeLevel,omitempty"` - Fee string `json:"fee,omitempty"` - PriorityFee string `json:"priorityFee,omitempty"` - MaxFee string `json:"maxFee,omitempty"` - GasLimit string `json:"gasLimit,omitempty"` - GasPrice string `json:"gasPrice,omitempty"` - NetworkFee string `json:"networkFee,omitempty"` - ReplaceTxByHash string `json:"replaceTxByHash,omitempty"` - ExtraParameters *extraParameters `json:"extraParameters,omitempty"` -} - -type source struct { - Type string `json:"type"` - SubType string `json:"subType,omitempty"` - ID string `json:"id,omitempty"` - Name string `json:"name,omitempty"` - WalletID string `json:"walletId,omitempty"` -} - -type destination struct { - Type string `json:"type"` - SubType string `json:"subType,omitempty"` - ID string `json:"id,omitempty"` - Name string `json:"name,omitempty"` - WalletID string `json:"walletId,omitempty"` - OneTimeAddress *oneTimeAddress `json:"oneTimeAddress,omitempty"` -} - -type oneTimeAddress struct { - Address string `json:"address,omitempty"` - Tag string `json:"tag,omitempty"` -} - -type extraParameters struct { - RawMessageData rawMessageData `json:"rawMessageData"` -} - -type rawMessageData struct { - Messages []unsignedRawMessage `json:"messages"` - Algorithm string `json:"algorithm,omitempty"` -} - -type unsignedRawMessage struct { - Content string `json:"content"` -} - -type transaction struct { - ID string `json:"id"` - ExternalTxID string `json:"externalTxId,omitempty"` - Status Status `json:"status"` - SubStatus string `json:"subStatus,omitempty"` - TxHash string `json:"txHash"` - Operation string `json:"operation"` - Note string `json:"note,omitempty"` - AssetID string `json:"assetId,omitempty"` - Source source `json:"source"` - SourceAddress string `json:"sourceAddress,omitempty"` - Tag string `json:"tag,omitempty"` - Destination *destination `json:"destination"` - Destinations []any `json:"destinations,omitempty"` - DestinationAddress string `json:"destinationAddress,omitempty"` - DestinationAddressDescription string `json:"destinationAddressDescription,omitempty"` - DestinationTag string `json:"destinationTag,omitempty"` - ContractCallDecodedData any `json:"contractCallDecodedData,omitempty"` - AmountInfo *amountInfo `json:"amountInfo,omitempty"` - TreatAsGrossAmount bool `json:"treatAsGrossAmount"` - FeeInfo *feeInfo `json:"feeInfo,omitempty"` - FeeCurrency string `json:"feeCurrency,omitempty"` - NetworkRecords []networkRecords `json:"networkRecords,omitempty"` - CreatedAt int `json:"createdAt"` - LastUpdated int `json:"lastUpdated"` - CreatedBy string `json:"createdBy"` - SignedBy []string `json:"signedBy"` - RejectedBy string `json:"rejectedBy"` - AuthorizationInfo *authorizationInfo `json:"authorizationInfo"` - ExchangeTxID string `json:"exchangeTxId,omitempty"` - CustomerRefID string `json:"customerRefId,omitempty"` - AmlScreeningResult *amlScreeningResult `json:"amlScreeningResult,omitempty"` - ExtraParameters map[string]any `json:"extraParameters,omitempty"` - SignedMessages []signedMessage `json:"signedMessages"` - NumOfConfirmations int `json:"numOfConfirmations"` - BlockInfo *blockInfo `json:"blockInfo"` - Index int `json:"index"` - RewardInfo *rewardInfo `json:"rewardInfo,omitempty"` - SystemMessages *systemMessages `json:"systemMessages,omitempty"` - AddressType string `json:"addressType"` -} - -// Pubkey0 returns the public key of the first signed message in the transaction. -func (t transaction) Pubkey0() (*ecdsa.PublicKey, error) { - if len(t.SignedMessages) != 1 { - return nil, errors.New("unexpected number of signed messages", "count", len(t.SignedMessages)) - } - - msg := t.SignedMessages[0] - - pk, err := hex.DecodeString(msg.PublicKey) - if err != nil { - return nil, errors.Wrap(err, "decode public key") - } - - pubkey, err := crypto.DecompressPubkey(pk) - if err != nil { - return nil, errors.Wrap(err, "decompress public key") - } - - return pubkey, nil -} - -// Sig0 returns the signature (Ethereum RSV format) of the first signed message in the transaction. -func (t transaction) Sig0() ([65]byte, error) { - if len(t.SignedMessages) != 1 { - return [65]byte{}, errors.New("unexpected number of signed messages", "count", len(t.SignedMessages)) - } - - msg := t.SignedMessages[0] - - // FullSig field is [R || S] in hex format. - sig, err := hex.DecodeString(msg.Signature.FullSig) - if err != nil { - return [65]byte{}, errors.Wrap(err, "decode signature") - } else if len(sig) != 64 { - return [65]byte{}, errors.New("unexpected signature length", "length", len(sig)) - } - - // V is either 0 or 1 - sig = append(sig, byte(msg.Signature.V)) - - return [65]byte(sig), nil -} - -type amountInfo struct { - Amount string `json:"amount,omitempty"` - RequestedAmount string `json:"requestedAmount,omitempty"` - NetAmount string `json:"netAmount,omitempty"` - AmountUSD string `json:"amountUSD,omitempty"` -} - -type feeInfo struct { - NetworkFee string `json:"networkFee,omitempty"` - ServiceFee string `json:"serviceFee,omitempty"` - GasPrice string `json:"gasPrice,omitempty"` -} - -type amlScreeningResult struct { - Provider string `json:"provider,omitempty"` - Payload any `json:"payload,omitempty"` -} - -type networkRecords struct { - Source source `json:"source"` - Destination *destination `json:"destination,omitempty"` - TxHash string `json:"txHash,omitempty"` - NetworkFee string `json:"networkFee,omitempty"` - AssetID string `json:"assetId,omitempty"` - NetAmount string `json:"netAmount,omitempty"` - IsDropped bool `json:"isDropped,omitempty"` - Type string `json:"type,omitempty"` - DestinationAddress string `json:"destinationAddress,omitempty"` - SourceAddress string `json:"sourceAddress,omitempty"` - AmountUSD string `json:"amountUSD,omitempty"` - Index int `json:"index"` - RewardInfo *rewardInfo `json:"rewardInfo,omitempty"` -} - -type rewardInfo struct { - SrcRewards string `json:"srcRewards"` - DestRewards string `json:"destRewards"` -} - -type signature struct { - FullSig string `json:"fullSig"` - R string `json:"r"` - S string `json:"s"` - V int `json:"v"` -} -type signedMessage struct { - Content string `json:"content"` - Algorithm string `json:"algorithm"` - DerivationPath []int `json:"derivationPath"` - Signature signature `json:"signature"` - PublicKey string `json:"publicKey"` -} -type blockInfo struct { - BlockHeight string `json:"blockHeight"` - BlockHash string `json:"blockHash"` -} - -type users struct { - AdditionalProp string `json:"additionalProp"` -} -type groups struct { - Th int `json:"th"` - Users users `json:"users"` -} -type authorizationInfo struct { - AllowOperatorAsAuthorizer bool `json:"allowOperatorAsAuthorizer"` - Logic string `json:"logic"` - Groups []groups `json:"groups"` -} - -type createTransactionResponse struct { - ID string `json:"id"` - Status string `json:"status"` - SystemMessages *systemMessages `json:"systemMessages,omitempty"` -} - -type systemMessages struct { - Type string `json:"type"` - Message string `json:"message"` -} - -type Asset struct { - ID string `json:"id"` - Name string `json:"name"` - Type string `json:"type"` - ContractAddress string `json:"contractAddress,omitempty"` - NativeAsset string `json:"nativeAsset,omitempty"` - Decimals int `json:"decimals,omitempty"` -} - -type pubkeyResponse struct { - Algorithm string `json:"algorithm"` - DerivationPath []int `json:"derivationPath"` - PublicKey string `json:"publicKey"` -} - -type errorResponse struct { - Message string `json:"message"` - Code int `json:"code"` -} - -type vaultsResponse struct { - Accounts []account - Paging paging -} - -type account struct { - ID string `json:"id"` - Name string `json:"name"` -} - -type paging struct { - Before string `json:"before"` - After string `json:"after"` -} diff --git a/lib/forkjoin/forkjoin.go b/lib/forkjoin/forkjoin.go deleted file mode 100644 index 7fa42eac..00000000 --- a/lib/forkjoin/forkjoin.go +++ /dev/null @@ -1,264 +0,0 @@ -// Package forkjoin provides an API for "doing work -// concurrently (fork) and then waiting for the results (join)". -// It is similar to errgroups, except that each "group" doesn't only return an error, but also a result. -// -// This package was copied from Obol's Charon repo. -// See https://github.com/ObolNetwork/charon/blob/main/app/forkjoin/forkjoin.go. -package forkjoin - -import ( - "context" - "sync" - - "github.com/piplabs/story/lib/errors" -) - -const ( - defaultWorkers = 8 - defaultInputBuf = 100 - defaultFailFast = true - defaultWaitOnCancel = false -) - -// Fork function enqueues the input to be processed asynchronously. -// Note Fork may block temporarily while the input buffer is full, see WithInputBuffer. -// Note Fork will panic if called after Join. -type Fork[I any] func(I) - -// Join function closes the input queue and returns the result channel. -// Note Fork will panic if called after Join. -// Note Join must only be called once, otherwise panics. -type Join[I, O any] func() Results[I, O] - -// Work defines the work function signature workers will call. -type Work[I, O any] func(ctx context.Context, input I) (output O, err error) - -// Results contains enqueued results. -type Results[I, O any] <-chan Result[I, O] - -// Result contains the input and resulting output from the work function. -type Result[I, O any] struct { - Input I - Output O - Err error -} - -// Flatten blocks and returns all the outputs when all completed and -// the first "real error". -// -// A real error is the error that triggered the fail fast, all subsequent -// results will contain context canceled errors. -func (r Results[I, O]) Flatten() ([]O, error) { - //nolint:prealloc // We don't know the length of the results. - var ( - ctxErr error - otherErr error - resp []O - ) - for result := range r { - resp = append(resp, result.Output) - - if result.Err == nil { - continue - } - - if errors.Is(result.Err, context.Canceled) && ctxErr == nil { - ctxErr = result.Err - } - if !errors.Is(result.Err, context.Canceled) && otherErr == nil { - otherErr = result.Err - } - } - - if otherErr != nil { - return resp, otherErr - } else if ctxErr != nil { - return resp, ctxErr - } - - return resp, nil -} - -type options struct { - inputBuf int - workers int - failFast bool - waitOnCancel bool -} - -type Option func(*options) - -// WithWaitOnCancel returns an option configuring a forkjoin to wait for all workers to return when canceling. -// The default behavior just cancels the worker context and closes the output channel without waiting -// for the workers to return. -func WithWaitOnCancel() Option { - return func(o *options) { - o.waitOnCancel = true - } -} - -// WithWorkers returns an option configuring a forkjoin with w number of workers. -func WithWorkers(w int) Option { - return func(o *options) { - o.workers = w - } -} - -// WithInputBuffer returns an option configuring a forkjoin with an input buffer -// of length i overriding the default of 100. -// Useful to prevent temporary blocking during calls to Fork if enqueuing more than 100 inputs. -func WithInputBuffer(i int) Option { - return func(o *options) { - o.inputBuf = i - } -} - -// WithoutFailFast returns an option configuring a forkjoin to not stop execution on any error. -func WithoutFailFast() Option { - return func(o *options) { - o.failFast = false - } -} - -// New returns fork, join, and cancel functions with generic input type I and output type O. -// It provides an API for "doing work concurrently (fork) and then waiting for the results (join)". -// -// It fails fast by default, stopping execution on any error. All active work function contexts -// are canceled and no further inputs are executed with remaining result errors being set -// to context canceled. See WithoutFailFast. -// -// Usage: -// -// var workFunc := func(ctx context.Context, input MyInput) (MyResult, error) { -// ... do work -// return result, nil -// } -// -// fork, join, cancel := forkjoin.New[MyInput,MyResult](ctx, workFunc) -// defer cancel() // Release any remaining resources. -// -// for _, in := range inputs { -// fork(in) // Note that calling fork AFTER join panics! -// } -// -// resultChan := join() -// -// // Either read results from the channel as they appear -// for result := range resultChan { ... } -// -// // Or block until all results are complete and flatten -// results, firstErr := resultChan.Flatten() -func New[I, O any](rootCtx context.Context, work Work[I, O], opts ...Option) (Fork[I], Join[I, O], context.CancelFunc) { - options := options{ - workers: defaultWorkers, - inputBuf: defaultInputBuf, - failFast: defaultFailFast, - waitOnCancel: defaultWaitOnCancel, - } - - for _, opt := range opts { - opt(&options) - } - - var ( - wg sync.WaitGroup - zero O - input = make(chan I, options.inputBuf) - results = make(chan Result[I, O]) - dropOutput = make(chan struct{}) - done = make(chan struct{}) - ) - - workCtx, cancelWorkers := context.WithCancel(rootCtx) - - // enqueue result asynchronously since results channel is unbuffered/blocking. - enqueue := func(in I, out O, err error) { - go func() { - select { - case results <- Result[I, O]{ - Input: in, - Output: out, - Err: err, - }: - case <-dropOutput: - // Dropping output. - } - wg.Done() - }() - } - - for range options.workers { // Start workers - go func() { - for in := range input { // Process all inputs (channel closed on Join) - if workCtx.Err() != nil { // Skip work if failed fast - enqueue(in, zero, workCtx.Err()) - continue - } - - out, err := work(workCtx, in) - if options.failFast && err != nil { // Maybe fail fast - cancelWorkers() - } - - enqueue(in, out, err) - } - }() - } - - // Fork enqueues inputs, keeping track of how many was enqueued. - fork := func(i I) { - var added bool - defer func() { - // Handle panic use-case as well as rootCtx done. - if !added { - wg.Done() - } - }() - - wg.Add(1) - select { - case input <- i: - added = true - case <-rootCtx.Done(): - } - } - - // Join returns the results channel that will contain all the results in the future. - // It also closes the input queue (causing subsequent calls Fork to panic) - // It also starts a shutdown goroutine that closes the results channel when processing completed - join := func() Results[I, O] { - close(input) - - go func() { - // Auto close result channel when done - wg.Wait() - close(results) - close(done) - }() - - return results - } - - // cancel, drop remaining results and cancel workers if not done already. - cancel := func() { - close(dropOutput) - cancelWorkers() - if options.waitOnCancel { - <-done - } - } - - return fork, join, cancel -} - -// NewWithInputs is a convenience function that calls New and then forks all the inputs -// returning the join result and a cancel function. -func NewWithInputs[I, O any](ctx context.Context, work Work[I, O], inputs []I, opts ...Option, -) (Results[I, O], context.CancelFunc) { - fork, join, cancel := New[I, O](ctx, work, opts...) - for _, input := range inputs { - fork(input) - } - - return join(), cancel -} diff --git a/lib/forkjoin/forkjoin_test.go b/lib/forkjoin/forkjoin_test.go deleted file mode 100644 index 8d725f50..00000000 --- a/lib/forkjoin/forkjoin_test.go +++ /dev/null @@ -1,142 +0,0 @@ -// Copyright © 2022-2023 Obol Labs Inc. Licensed under the terms of a Business Source License 1.1 - -//nolint:paralleltest // Goleak doesn't support parrellel tests -package forkjoin_test - -import ( - "context" - "sort" - "testing" - - "github.com/stretchr/testify/require" - - "github.com/piplabs/story/lib/errors" - "github.com/piplabs/story/lib/forkjoin" - - "go.uber.org/goleak" -) - -func TestForkJoin(t *testing.T) { - ctx := context.Background() - - const n = 100 - testErr := errors.New("test error") - - tests := []struct { - name string - work forkjoin.Work[int, int] - failfast bool - expectedErr error - allOutput bool - }{ - { - name: "happy", - expectedErr: nil, - work: func(_ context.Context, i int) (int, error) { return i, nil }, - allOutput: true, - }, - { - name: "first error fast fail", - expectedErr: testErr, - failfast: true, - work: func(ctx context.Context, i int) (int, error) { - if i == 0 { - return 0, testErr - } - if i > n/2 { - require.Fail(t, "not failed fast") - } - <-ctx.Done() // This will hang if not failing fast - - return 0, ctx.Err() - }, - }, - { - name: "all error no fast fail", - allOutput: true, - expectedErr: testErr, - work: func(_ context.Context, i int) (int, error) { - return i, testErr - }, - }, - { - name: "all context cancel", - expectedErr: context.Canceled, - failfast: true, - work: func(_ context.Context, i int) (int, error) { - if i < n/2 { - return 0, context.Canceled - } - - return 0, nil - }, - }, - } - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - defer goleak.VerifyNone(t) - - var opts []forkjoin.Option - if !test.failfast { - opts = append(opts, forkjoin.WithoutFailFast()) - } - - fork, join, cancel := forkjoin.New[int, int](ctx, test.work, opts...) - defer cancel() - - var allOutput []int - for i := range n { - fork(i) - allOutput = append(allOutput, i) - } - - resp, err := join().Flatten() - require.Len(t, resp, n) - - if test.expectedErr != nil { - require.Equal(t, test.expectedErr, err) - } else { - require.NoError(t, err) - } - - if test.allOutput { - sort.Ints(resp) - require.Equal(t, allOutput, resp) - } - }) - } -} - -func TestPanic(t *testing.T) { - defer goleak.VerifyNone(t) - - fork, join, cancel := forkjoin.New[int, int](context.Background(), nil, forkjoin.WithWaitOnCancel()) - join() - cancel() - - // Calling fork after join panics - require.Panics(t, func() { - fork(0) - }) - - // Calling join again panics - require.Panics(t, func() { - join() - }) -} - -func TestLeak(t *testing.T) { - defer goleak.VerifyNone(t) - - fork, join, cancel := forkjoin.New[int, int]( - context.Background(), - func(ctx context.Context, i int) (int, error) { return i, nil }, - forkjoin.WithWaitOnCancel(), - ) - fork(1) - fork(2) - results := join() - <-results // Read 1 or 2 - cancel() // Fails if not called. -} diff --git a/lib/merkle/README.md b/lib/merkle/README.md deleted file mode 100644 index ccb6ae5d..00000000 --- a/lib/merkle/README.md +++ /dev/null @@ -1,5 +0,0 @@ -# merkle - -This is a port of OpenZeppelin's JS [merkle-tree](https://github.com/OpenZeppelin/merkle-tree/tree/master) library to Go. -It excludes the `StandardTree` wrapper types since our use-case doesn't require it. -It also moves all non-iliad-required logic to the test package to decrease the surface area of the library. diff --git a/lib/merkle/core.go b/lib/merkle/core.go deleted file mode 100644 index a0648db2..00000000 --- a/lib/merkle/core.go +++ /dev/null @@ -1,198 +0,0 @@ -// Package merkle provides an API to generate merkle trees and proofs from 32 byte leaves. -// It is a port of the OpenZeppelin JS merkle-tree library. -// See https://github.com/OpenZeppelin/merkle-tree/tree/master. -package merkle - -import ( - "bytes" - "sort" - - "github.com/ethereum/go-ethereum/crypto" - - "github.com/piplabs/story/lib/errors" -) - -// StdLeafHash returns the standard leaf hash of the given data. -// The data is hashed twice with keccak256 to prevent pre-image attacks. -func StdLeafHash(data []byte) [32]byte { - h1 := hash(data) - h2 := hash(h1[:]) - - return h2 -} - -// MakeTree returns a merkle tree given the leaves. -func MakeTree(leaves [][32]byte) ([][32]byte, error) { - if len(leaves) == 0 { - return nil, errors.New("no leaves provided") - } - - treeLen := 2*len(leaves) - 1 - tree := make([][32]byte, treeLen) - - // Fill in leaves in reverse order. - for i, leaf := range leaves { - tree[treeLen-1-i] = leaf - } - - // Fill in the intermediate nodes up to the root. - for i := treeLen - 1 - len(leaves); i >= 0; i-- { - tree[i] = hashPair(tree[leftChildIndex(i)], tree[rightChildIndex(i)]) - } - - return tree, nil -} - -// MultiProof is a merkle-multi-proof for multiple leaves. -type MultiProof struct { - Leaves [][32]byte - Proof [][32]byte - ProofFlags []bool -} - -// GetMultiProof returns a merkle-multi-proof for the given leaf indices. -func GetMultiProof(tree [][32]byte, indices ...int) (MultiProof, error) { - if len(indices) == 0 { - return MultiProof{}, errors.New("no indices provided") - } - - for _, i := range indices { - if err := checkLeafNode(tree, i); err != nil { - return MultiProof{}, err - } - } - - // Sort indices in reverse order. - sort.Slice(indices, func(i, j int) bool { - return indices[i] > indices[j] - }) - - // Check for duplicates. - for i, j := range indices[1:] { - if j == indices[i] { - return MultiProof{}, errors.New("cannot prove duplicated index") - } - } - - stack := make([]int, len(indices)) - copy(stack, indices) - var proof [][32]byte - var proofFlags []bool - - for len(stack) > 0 && stack[0] > 0 { - // Pop from the beginning. - j := stack[0] - stack = stack[1:] - - s := siblingIndex(j) - p := parentIndex(j) - - if len(stack) > 0 && s == stack[0] { - proofFlags = append(proofFlags, true) - stack = stack[1:] - } else { - proofFlags = append(proofFlags, false) - proof = append(proof, tree[s]) - } - stack = append(stack, p) //nolint:makezero // Appending to non-zero initialized slice is ok - } - - leaves := make([][32]byte, 0, len(indices)) - for _, i := range indices { - leaves = append(leaves, tree[i]) - } - - return MultiProof{ - Leaves: leaves, - Proof: proof, - ProofFlags: proofFlags, - }, nil -} - -// isTreeNode returns true if the given index is a node in the tree. -func isTreeNode(tree [][32]byte, i int) bool { - return i >= 0 && i < len(tree) -} - -// isInternalNode returns true if the given index is an internal node (not a leaf) in the tree. -func isInternalNode(tree [][32]byte, i int) bool { - return isTreeNode(tree, leftChildIndex(i)) -} - -// isLeafNode returns true if the given index is a leaf node in the tree. -func isLeafNode(tree [][32]byte, i int) bool { - return isTreeNode(tree, i) && !isInternalNode(tree, i) -} - -// checkLeafNode returns an error if the given index is not a leaf node. -func checkLeafNode(tree [][32]byte, i int) error { - if !isLeafNode(tree, i) { - return errors.New("index is not a leaf") - } - - return nil -} - -// leftChildIndex returns the index of the left child of the node at the given index. -func leftChildIndex(i int) int { - return 2*i + 1 -} - -// rightChildIndex returns the index of the right child of the node at the given index. -func rightChildIndex(i int) int { - return 2*i + 2 -} - -// parentIndex returns the index of the parent of the node at the given index. -// It panics if the given index is 0. -func parentIndex(i int) int { - if i == 0 { - panic("root has no parent") - } - - return (i - 1) / 2 -} - -// siblingIndex returns the index of the sibling of the node at the given index. -func siblingIndex(i int) int { - if i == 0 { - panic("root has no sibling") - } - - if i%2 == 0 { - return i - 1 - } - - return i + 1 -} - -// sortBytes returns the given byte slices sorted in ascending order. -func sortBytes(buf ...[]byte) [][]byte { - sort.Slice(buf, func(i, j int) bool { - return bytes.Compare(buf[i], buf[j]) < 0 - }) - - return buf -} - -// concatBytes returns the concatenation of the given byte slices. -func concatBytes(buf ...[]byte) []byte { - var resp []byte - for _, b := range buf { - resp = append(resp, b...) - } - - return resp -} - -// hashPair returns the 32 byte keccak256 hash of the sorted concatenation of the given byte arrays. -func hashPair(a [32]byte, b [32]byte) [32]byte { - return hash(concatBytes(sortBytes(a[:], b[:])...)) -} - -// hash returns the 32 byte keccak256 hash of the given byte slice. -func hash(buf []byte) [32]byte { - resp := crypto.Keccak256(buf) - - return [32]byte(resp) -} diff --git a/lib/merkle/core_internal_test.go b/lib/merkle/core_internal_test.go deleted file mode 100644 index 245c7461..00000000 --- a/lib/merkle/core_internal_test.go +++ /dev/null @@ -1,102 +0,0 @@ -package merkle - -import "github.com/piplabs/story/lib/errors" - -// These functions are also ported from OpenZeppelin's library, but they -// are not used by iliad's production code, so they are part of the -// tests to decrease prod code surface. - -// GetProof returns a merkle proof for the given leaf index. -func GetProof(tree [][32]byte, index int) ([][32]byte, error) { - if err := checkLeafNode(tree, index); err != nil { - return nil, err - } - - var proof [][32]byte - for index > 0 { - proof = append(proof, tree[siblingIndex(index)]) - index = parentIndex(index) - } - - return proof, nil -} - -// ProcessProof returns the root hash of the merkle tree given the leaf hash and the proof. -func ProcessProof(leaf [32]byte, proof [][32]byte) [32]byte { - node := leaf - for _, p := range proof { - node = hashPair(node, p) - } - - return node -} - -// ProcessMultiProof returns the root hash of the tree given a multi proof. -func ProcessMultiProof(multi MultiProof) ([32]byte, error) { - if err := verifyMultiProof(multi); err != nil { - return [32]byte{}, err - } - - // Copy leaves and proof. - stack := make([][32]byte, len(multi.Leaves)) - copy(stack, multi.Leaves) - proof := make([][32]byte, len(multi.Proof)) - copy(proof, multi.Proof) - - for _, flag := range multi.ProofFlags { - // Pop from the beginning of the stack. - a := stack[0] - stack = stack[1:] - - // Either pop from the stack or the proof, depending on the flag. - var b [32]byte - if flag { - b = stack[0] - stack = stack[1:] - } else { - b = proof[0] - proof = proof[1:] - } - - stack = append(stack, hashPair(a, b)) //nolint:makezero // Appending to non-zero initialized slice is ok - } - - // Either the stack or the proof should have one element left. - if len(stack)+len(proof) != 1 { - return [32]byte{}, errors.New("broken invariant") - } - - if len(stack) > 0 { - return stack[0], nil - } - - return proof[0], nil -} - -// LeafToTreeIndex returns the index of the leaf in the tree given the original index in the leaves slice. -func LeafToTreeIndex(tree [][32]byte, leafIndex int) int { - return len(tree) - 1 - leafIndex -} - -// verifyMultiProof returns an error if the given multi proof is invalid. -func verifyMultiProof(multi MultiProof) error { - var falseFlags int - for _, flag := range multi.ProofFlags { - if !flag { - falseFlags++ - } - } - if len(multi.Proof) != falseFlags { - return errors.New("false proof flags don't match proof") - } - - if len(multi.Leaves)+len(multi.Proof) != len(multi.ProofFlags)+1 { - return errors.New("proof flags don't match leaves and proof") - } - - if len(multi.Leaves) == 0 { - return errors.New("no leaves provided") - } - - return nil -} diff --git a/lib/merkle/core_test.go b/lib/merkle/core_test.go deleted file mode 100644 index e1d23c06..00000000 --- a/lib/merkle/core_test.go +++ /dev/null @@ -1,87 +0,0 @@ -package merkle_test - -import ( - "math/rand" - "testing" - - fuzz "github.com/google/gofuzz" - "github.com/stretchr/testify/require" - - "github.com/piplabs/story/lib/merkle" -) - -// TestLeaveProvable tests that a leaf can be proven. -func TestLeaveProvable(t *testing.T) { - t.Parallel() - - // Create random leaves - var leaves [][32]byte - fuzz.New().NilChance(0).NumElements(1, 256).Fuzz(&leaves) - - // Make tree - tree, err := merkle.MakeTree(leaves) - require.NoError(t, err) - - // Pick random leaf - leafIndex := rand.Intn(len(leaves)) - treeIndex := merkle.LeafToTreeIndex(tree, leafIndex) - - // Get the proof - proof, err := merkle.GetProof(tree, treeIndex) - require.NoError(t, err) - - leaf := leaves[leafIndex] - root := merkle.ProcessProof(leaf, proof) - require.Equal(t, tree[0], root) -} - -// TestLeavesProvable tests that multiple leaves can be proven. -func TestLeavesProvable(t *testing.T) { - t.Parallel() - - // Create random leaves - var leaves [][32]byte - fuzz.New().NilChance(0).NumElements(1, 256).Fuzz(&leaves) - - // Make tree - tree, err := merkle.MakeTree(leaves) - require.NoError(t, err) - - // Pick random leaves - leafIndices := randomIndicesRange(leaves) - treeIndices := make([]int, len(leafIndices)) - for i, leafIndex := range leafIndices { - treeIndices[i] = merkle.LeafToTreeIndex(tree, leafIndex) - } - - // Get the multi proof - multi, err := merkle.GetMultiProof(tree, treeIndices...) - require.NoError(t, err) - - // Check that the proof contains the leaves - require.Equal(t, len(leafIndices), len(multi.Leaves)) - for _, i := range leafIndices { - require.Contains(t, multi.Leaves, leaves[i]) - } - - // Check that the proof is valid - root, err := merkle.ProcessMultiProof(multi) - require.NoError(t, err) - require.Equal(t, tree[0], root) -} - -// randomIndicesRange returns a random range of indices of the provided slice. -func randomIndicesRange(slice [][32]byte) []int { - start := rand.Intn(len(slice)) - count := rand.Intn(len(slice) - start) - if count == 0 { - count = 1 - } - - indices := make([]int, count) - for i := range indices { - indices[i] = start + i - } - - return indices -} diff --git a/lib/stream/stream.go b/lib/stream/stream.go deleted file mode 100644 index 0079b961..00000000 --- a/lib/stream/stream.go +++ /dev/null @@ -1,341 +0,0 @@ -// Package stream provide a generic stream function. -package stream - -import ( - "context" - "sync" - "time" - - "github.com/piplabs/story/lib/errors" - "github.com/piplabs/story/lib/log" - - "go.opentelemetry.io/otel/trace" -) - -type Callback[E any] func(ctx context.Context, elem E) error - -type Deps[E any] struct { - // Dependency functions - - // FetchBatch fetches the next batch of elements from the provided height (inclusive). - // The elements must be sequential, since the internal height cursors is incremented for each element returned. - FetchBatch func(ctx context.Context, chainID uint64, height uint64) ([]E, error) - // Backoff returns a backoff function. See expbackoff package for the implementation. - Backoff func(ctx context.Context) func() - // Verify is a sanity check function, it ensures each element is valid. - Verify func(ctx context.Context, elem E, height uint64) error - // Height returns the height of an element. - Height func(elem E) uint64 - - // Config - FetchWorkers uint64 - ElemLabel string - RetryCallback bool - - // Metrics - IncFetchErr func() - IncCallbackErr func() - SetStreamHeight func(uint64) - SetCallbackLatency func(time.Duration) - StartTrace func(ctx context.Context, height uint64, spanName string) (context.Context, trace.Span) -} - -// Stream streams elements from the provided height (inclusive) of a specific chain. -// It fetches the batches of elements from the current height, and -// calls the callback function for each element in strictly-sequential order. -// -// It supports concurrent fetching of single-element-batches only. -// It retries forever on fetch errors. -// It can either retry or return callback errors. -// It returns (nil) when the context is canceled. -// - -func Stream[E any](ctx context.Context, deps Deps[E], srcChainID uint64, startHeight uint64, callback Callback[E]) error { - if deps.FetchWorkers == 0 { - return errors.New("invalid zero fetch worker count") - } - - // Define a robust fetch function that fetches a batch of elements from a height (inclusive). - // It only returns an empty list if the context is canceled. - // It retries forever on error or if no elements found. - fetchFunc := func(ctx context.Context, height uint64) []E { - backoff := deps.Backoff(ctx) // Note that backoff returns immediately on ctx cancel. - for { - if ctx.Err() != nil { - return nil - } - - fetchCtx, span := deps.StartTrace(ctx, height, "fetch") - elems, err := deps.FetchBatch(fetchCtx, srcChainID, height) - span.End() - - if ctx.Err() != nil { - return nil - } else if err != nil { - log.Warn(ctx, "Failed fetching "+deps.ElemLabel+" (will retry)", err, "height", height) - deps.IncFetchErr() - backoff() - - continue - } else if len(elems) == 0 { - // We reached the head of the chain, wait for new blocks. - backoff() - - continue - } - - heightsOK := true - for i, elem := range elems { - if h := deps.Height(elem); h != height+uint64(i) { - log.Error(ctx, "Invalid "+deps.ElemLabel+" height [BUG]", nil, - "expect", height, - "actual", h, - ) - - heightsOK = false - } - } - if !heightsOK { // Can't return invalid elements, just retry fetching for now. - backoff() - continue - } - - return elems - } - } - - // Define a robust callback function that retries on error. - callbackFunc := func(ctx context.Context, elem E) error { - height := deps.Height(elem) - ctx, span := deps.StartTrace(ctx, height, "callback") - defer span.End() - ctx = log.WithCtx(ctx, "height", height) - - backoff := deps.Backoff(ctx) - - if err := deps.Verify(ctx, elem, height); err != nil { - return errors.Wrap(err, "verify") - } - - // Retry callback on error - for { - if ctx.Err() != nil { - return nil // Don't backoff or log on ctx cancel, just return nil. - } - - t0 := time.Now() - err := callback(ctx, elem) - deps.SetCallbackLatency(time.Since(t0)) - if ctx.Err() != nil { - return nil // Don't backoff or log on ctx cancel, just return nil. - } else if err != nil && !deps.RetryCallback { - deps.IncCallbackErr() - return errors.Wrap(err, "callback") - } else if err != nil { - log.Warn(ctx, "Failed processing "+deps.ElemLabel+" (will retry)", err) - deps.IncCallbackErr() - backoff() - - continue - } - - deps.SetStreamHeight(height) - - return nil - } - } - - // Sorting buffer connects the concurrent fetch workers to the callback - sorter := newSortingBuffer(startHeight, deps, callbackFunc) - - // Start fetching workers - startFetchWorkers(ctx, deps, fetchFunc, sorter, startHeight) - - // Sort fetch results and call callback - return sorter.Process(ctx) -} - -// startFetchWorkers starts worker goroutines -// that fetch all batches concurrently from the provided . -// -// Concurrent fetching is only supported for single-element-batches, since -// each worker fetches: startHeight + Ith-iteration + Nth-worker. -// -// For multi-element-batches, only a single worker is supported. -func startFetchWorkers[E any]( - ctx context.Context, - deps Deps[E], - work func(ctx context.Context, height uint64) []E, - sorter *sortingBuffer[E], - startHeight uint64, -) { - for i := range deps.FetchWorkers { - go func(workerID int, height uint64) { - for { - // Work function MUST be robust, always returning a non-empty strictly-sequential batch - // or nil if the context was canceled. - batch := work(ctx, height) - if ctx.Err() != nil { - return - } else if len(batch) == 0 { - log.Error(ctx, "Work function returned an empty batch [BUG]", nil) - return - } else if len(batch) > 1 && deps.FetchWorkers > 1 { - log.Error(ctx, "Concurrent fetching only supported for single element batches [BUG]", nil) - return - } - - var last uint64 - for i, e := range batch { - last = deps.Height(e) - if last != height+uint64(i) { - log.Error(ctx, "Invalid batch [BUG]", nil) - return - } - } - - sorter.Add(ctx, workerID, batch) - - // Calculate next height to fetch - height = last + deps.FetchWorkers - } - }(int(i), startHeight+i) // Initialize a height to fetch per worker - } -} - -// sortingBuffer buffers unordered batches of elements (one batch per worker), -// providing elements to the callback in strictly-sequential sorted order. -type sortingBuffer[E any] struct { - deps Deps[E] - callback func(ctx context.Context, elem E) error - startHeight uint64 - - mu sync.Mutex - buffer map[uint64]workerElem[E] // Worker elements by height - counts map[int]int // Count of elements per worker - signals map[int]chan struct{} // Processes <> Worker comms -} - -const processorID = -1 - -func newSortingBuffer[E any]( - startHeight uint64, - deps Deps[E], - callback func(ctx context.Context, elem E) error, -) *sortingBuffer[E] { - signals := make(map[int]chan struct{}) - signals[processorID] = make(chan struct{}, 1) - for i := range int(deps.FetchWorkers) { - signals[i] = make(chan struct{}, 1) - } - - return &sortingBuffer[E]{ - startHeight: startHeight, - deps: deps, - callback: callback, - buffer: make(map[uint64]workerElem[E]), - counts: make(map[int]int), - signals: signals, - } -} - -// signal signals the ID to wakeup. -func (m *sortingBuffer[E]) signal(signalID int) { - select { - case m.signals[signalID] <- struct{}{}: - default: - } -} - -// retryLock repeatedly obtains the lock and calls the callback while it returns false. -// It returns once the callback returns true or an error. -func (m *sortingBuffer[E]) retryLock(ctx context.Context, signalID int, fn func(ctx context.Context) (bool, error)) error { - timer := time.NewTicker(time.Nanosecond) // Initial timer is instant - defer timer.Stop() - - for { - select { - case <-ctx.Done(): - return nil - case <-m.signals[signalID]: - case <-timer.C: - } - - m.mu.Lock() - done, err := fn(ctx) - m.mu.Unlock() - if err != nil { - return err - } else if done { - return nil - } - - // Not done, so retry again, much later - timer.Reset(time.Second) - } -} - -func (m *sortingBuffer[E]) Add(ctx context.Context, workerID int, batch []E) { - _ = m.retryLock(ctx, workerID, func(_ context.Context) (bool, error) { - // Wait for any previous batch this worker added to be processed before adding this batch. - // This results in backpressure to workers, basically only buffering a single batch per worker. - if m.counts[workerID] > 0 { - return false, nil // Previous batch still in buffer, retry a bit later - } - - // Add the batch - for _, e := range batch { - height := m.deps.Height(e) - - // Invariant check (error handling in workerFunc) - if _, ok := m.buffer[height]; ok { - return false, errors.New("duplicate element [BUG]") - } - - m.buffer[height] = workerElem[E]{WorkerID: workerID, E: e} - } - - m.counts[workerID] = len(batch) - m.signal(processorID) // Signal the processor - - return true, nil // Don't retry lock again, we are done. - }) -} - -// Process calls the callback function in strictly-sequential order from (inclusive) -// as elements become available in the buffer. -func (m *sortingBuffer[E]) Process(ctx context.Context) error { - next := m.startHeight - return m.retryLock(ctx, processorID, func(ctx context.Context) (bool, error) { - elem, ok := m.buffer[next] - if !ok { - return false, nil // Next height not in buffer, retry a bit later - } - delete(m.buffer, next) - - err := m.callback(ctx, elem.E) - if err != nil { - return false, err // Don't retry again - } - - m.counts[elem.WorkerID]-- - - if m.counts[elem.WorkerID] == 0 { - m.signal(elem.WorkerID) // Signal the worker that it can add another batch - } - - next++ - - if _, ok := m.buffer[next]; ok { - m.signal(processorID) // Signal ourselves if next elements already in buffer. - } - - return false, nil // Retry again with next height - }) -} - -// workerElem represents an element processed by a worker. -type workerElem[E any] struct { - WorkerID int - E E -} diff --git a/lib/tokens/coingecko/coingecko.go b/lib/tokens/coingecko/coingecko.go deleted file mode 100644 index 3c7f219d..00000000 --- a/lib/tokens/coingecko/coingecko.go +++ /dev/null @@ -1,107 +0,0 @@ -package coingecko - -import ( - "context" - "encoding/json" - "net/http" - "net/url" - "strings" - - "github.com/piplabs/story/lib/errors" - "github.com/piplabs/story/lib/tokens" -) - -const ( - endpointSimplePrice = "/api/v3/simple/price" - prodHost = "https://api.coingecko.com" -) - -type Client struct { - host string -} - -var _ tokens.Pricer = Client{} - -// New creates a new goingecko Client with the given options. -func New(opts ...func(*options)) Client { - o := defaultOpts() - for _, opt := range opts { - opt(&o) - } - - return Client{ - host: o.Host, - } -} - -// GetPriceUSD returns the price of each coin in USD. -func (c Client) Price(ctx context.Context, tkns ...tokens.Token) (map[tokens.Token]float64, error) { - return c.getPrice(ctx, "usd", tkns...) -} - -// simplePriceResponse is the response from the simple/price endpoint. -// It mapes coin id to currency to price. -type simplePriceResponse map[string]map[string]float64 - -// GetPrice returns the price of each coin in the given currency. -func (c Client) getPrice(ctx context.Context, currency string, tkns ...tokens.Token) (map[tokens.Token]float64, error) { - ids := make([]string, len(tkns)) - for i, t := range tkns { - ids[i] = t.CoingeckoID() - } - - params := url.Values{ - "ids": {strings.Join(ids, ",")}, - "vs_currencies": {currency}, - } - - var resp simplePriceResponse - if err := c.doReq(ctx, endpointSimplePrice, params, &resp); err != nil { - return nil, err - } - - prices := make(map[tokens.Token]float64) - for id, price := range resp { - prices[tokens.MustFromCoingeckoID(id)] = price[currency] - } - - return prices, nil -} - -// doReq makes a GET request to the given path & params, and decodes the response into response. -func (c Client) doReq(ctx context.Context, path string, params url.Values, response any) error { - uri, err := c.uri(path, params) - if err != nil { - return err - } - - req, err := http.NewRequestWithContext(ctx, http.MethodGet, uri.String(), nil) - if err != nil { - return errors.Wrap(err, "create request", "url", uri.String()) - } - - resp, err := http.DefaultClient.Do(req) - if err != nil { - return errors.Wrap(err, "get", "url", uri.String()) - } - - defer resp.Body.Close() - if resp.StatusCode != http.StatusOK { - return errors.New("get", "url", uri.String(), "status", resp.Status) - } - - if err := json.NewDecoder(resp.Body).Decode(response); err != nil { - return errors.Wrap(err, "decode response") - } - - return nil -} - -func (c Client) uri(path string, params url.Values) (*url.URL, error) { - uri, err := url.Parse(c.host + path + "?" + params.Encode()) - if err != nil { - return nil, errors.Wrap(err, "parse url", "host", c.host, "path", path, "params", params.Encode()) - } - - return uri, nil -} diff --git a/lib/tokens/coingecko/coingecko_test.go b/lib/tokens/coingecko/coingecko_test.go deleted file mode 100644 index 0121388d..00000000 --- a/lib/tokens/coingecko/coingecko_test.go +++ /dev/null @@ -1,65 +0,0 @@ -package coingecko_test - -import ( - "context" - "encoding/json" - "math/rand" - "net/http" - "net/http/httptest" - "strings" - "testing" - - "github.com/stretchr/testify/require" - - "github.com/piplabs/story/lib/tokens" - "github.com/piplabs/story/lib/tokens/coingecko" - - "gotest.tools/v3/assert" -) - -func TestGetPrice(t *testing.T) { - t.Parallel() - - // map token id -> currency -> price - // set during request handler - testPrices := make(map[string]map[string]float64) - - ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - assert.Equal(t, "/api/v3/simple/price", r.URL.Path) - - q := r.URL.Query() - ids := strings.Split(q.Get("ids"), ",") - currencies := strings.Split(q.Get("vs_currencies"), ",") - - resp := make(map[string]map[string]float64) - for _, id := range ids { - resp[id] = make(map[string]float64) - - if _, ok := testPrices[id]; !ok { - testPrices[id] = make(map[string]float64) - } - - for _, currency := range currencies { - resp[id][currency] = randPrice() - - // also store the price, so we can assert against it - testPrices[id][currency] = resp[id][currency] - } - } - - bz, _ := json.Marshal(resp) - _, _ = w.Write(bz) - })) - - defer ts.Close() - - c := coingecko.New(coingecko.WithHost(ts.URL)) - prices, err := c.Price(context.Background(), tokens.ILIAD, tokens.ETH) - require.NoError(t, err) - require.InEpsilon(t, prices[tokens.ILIAD], testPrices[tokens.ILIAD.CoingeckoID()]["usd"], 0.01) - require.InEpsilon(t, prices[tokens.ETH], testPrices[tokens.ETH.CoingeckoID()]["usd"], 0.01) -} - -func randPrice() float64 { - return float64(int(rand.Float64()*10000)) / 100 -} diff --git a/lib/tokens/coingecko/options.go b/lib/tokens/coingecko/options.go deleted file mode 100644 index 55021a06..00000000 --- a/lib/tokens/coingecko/options.go +++ /dev/null @@ -1,17 +0,0 @@ -package coingecko - -type options struct { - Host string -} - -func WithHost(host string) func(*options) { - return func(o *options) { - o.Host = host - } -} - -func defaultOpts() options { - return options{ - Host: prodHost, - } -} diff --git a/lib/tokens/mock.go b/lib/tokens/mock.go deleted file mode 100644 index 198941d2..00000000 --- a/lib/tokens/mock.go +++ /dev/null @@ -1,41 +0,0 @@ -package tokens - -import ( - "context" - "sync" -) - -type MockPricer struct { - mu sync.RWMutex - prices map[Token]float64 -} - -var _ Pricer = (*MockPricer)(nil) - -func NewMockPricer(prices map[Token]float64) *MockPricer { - cloned := make(map[Token]float64) - for k, v := range prices { - cloned[k] = v - } - - return &MockPricer{prices: cloned} -} - -func (m *MockPricer) Price(_ context.Context, tkns ...Token) (map[Token]float64, error) { - m.mu.RLock() - defer m.mu.RUnlock() - - resp := make(map[Token]float64) - for _, t := range tkns { - resp[t] = m.prices[t] - } - - return resp, nil -} - -func (m *MockPricer) SetPrice(token Token, price float64) { - m.mu.Lock() - defer m.mu.Unlock() - - m.prices[token] = price -} diff --git a/lib/tokens/price.go b/lib/tokens/price.go deleted file mode 100644 index d01ffeaf..00000000 --- a/lib/tokens/price.go +++ /dev/null @@ -1,57 +0,0 @@ -package tokens - -import ( - "context" -) - -// Pricer is the token price provider interface. -type Pricer interface { - // Price returns the price of each provided token in USD. - Price(ctx context.Context, tokens ...Token) (map[Token]float64, error) -} - -type CachedPricer struct { - p Pricer - cache map[Token]float64 -} - -func NewCachedPricer(p Pricer) *CachedPricer { - return &CachedPricer{ - p: p, - cache: make(map[Token]float64), - } -} - -func (c *CachedPricer) Price(ctx context.Context, tokens ...Token) (map[Token]float64, error) { - prices := make(map[Token]float64) - - var uncached []Token - - for _, token := range tokens { - if price, ok := c.cache[token]; ok { - prices[token] = price - } else { - uncached = append(uncached, token) - } - } - - if len(uncached) == 0 { - return prices, nil - } - - newPrices, err := c.p.Price(ctx, uncached...) - if err != nil { - return nil, err - } - - for token, price := range newPrices { - prices[token] = price - c.cache[token] = price - } - - return prices, nil -} - -func (c *CachedPricer) ClearCache() { - c.cache = make(map[Token]float64) -} diff --git a/lib/tokens/tokens.go b/lib/tokens/tokens.go deleted file mode 100644 index 5a3c94af..00000000 --- a/lib/tokens/tokens.go +++ /dev/null @@ -1,42 +0,0 @@ -package tokens - -type Token string - -const ( - ILIAD Token = "ILIAD" - ETH Token = "ETH" -) - -var ( - coingeckoIDs = map[Token]string{ - ILIAD: "storyprotocol", - ETH: "ethereum", - } -) - -func (t Token) String() string { - return string(t) -} - -func (t Token) CoingeckoID() string { - return coingeckoIDs[t] -} - -func FromCoingeckoID(id string) (Token, bool) { - for t, i := range coingeckoIDs { - if i == id { - return t, true - } - } - - return "", false -} - -func MustFromCoingeckoID(id string) Token { - t, ok := FromCoingeckoID(id) - if !ok { - panic("unknown coingecko id: " + id) - } - - return t -} From 34ed5a5ee27520c05e3c80b4e1d36978100f9dab Mon Sep 17 00:00:00 2001 From: Ze Date: Wed, 21 Aug 2024 16:12:11 -0700 Subject: [PATCH 2/2] fix lint --- go.mod | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/go.mod b/go.mod index cc16f3bf..9f6470fd 100644 --- a/go.mod +++ b/go.mod @@ -39,7 +39,6 @@ require ( github.com/syndtr/goleveldb v1.0.1 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.27.0 go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.27.0 - go.uber.org/goleak v1.3.0 go.uber.org/mock v0.4.0 golang.org/x/sync v0.7.0 // indirect golang.org/x/tools v0.21.0 @@ -244,7 +243,7 @@ require ( google.golang.org/genproto/googleapis/api v0.0.0-20240520151616-dc85e6b867a5 google.golang.org/genproto/googleapis/rpc v0.0.0-20240515191416-fc5f0ca64291 // indirect gopkg.in/ini.v1 v1.67.0 // indirect - gotest.tools/v3 v3.5.1 + gotest.tools/v3 v3.5.1 // indirect nhooyr.io/websocket v1.8.6 // indirect pgregory.net/rapid v1.1.0 // indirect rsc.io/tmplfunc v0.0.3 // indirect