Implement endpoint to list off-chain data by hashes

client/client.go

@@ -18,6 +18,7 @@ type Factory interface {
 // Client is the interface that defines the implementation of all the endpoints
 type Client interface {
 	GetOffChainData(ctx context.Context, hash common.Hash) ([]byte, error)
+	ListOffChainData(ctx context.Context, hashes []common.Hash) (map[common.Hash][]byte, error)
 	SignSequence(signedSequence types.SignedSequence) ([]byte, error)
 }
 
@@ -84,3 +85,27 @@ func (c *client) GetOffChainData(ctx context.Context, hash common.Hash) ([]byte,
 
 	return result, nil
 }
+
+// ListOffChainData returns data based on the given hashes
+func (c *client) ListOffChainData(ctx context.Context, hashes []common.Hash) (map[common.Hash][]byte, error) {
+	response, err := rpc.JSONRPCCallWithContext(ctx, c.url, "sync_listOffChainData", hashes)
+	if err != nil {
+		return nil, err
+	}
+
+	if response.Error != nil {
+		return nil, fmt.Errorf("%v %v", response.Error.Code, response.Error.Message)
+	}
+
+	result := make(map[common.Hash]types.ArgBytes)
+	if err = json.Unmarshal(response.Result, &result); err != nil {
+		return nil, err
+	}
+
+	preparedResult := make(map[common.Hash][]byte)
+	for key, val := range result {
+		preparedResult[key] = val
+	}
+
+	return preparedResult, nil
+}

client/client_test.go

@@ -169,3 +169,81 @@ func TestClient_GetOffChainData(t *testing.T) {
 		})
 	}
 }
+
+func TestClient_ListOffChainData(t *testing.T) {
+	tests := []struct {
+		name       string
+		hashes     []common.Hash
+		result     string
+		data       map[common.Hash][]byte
+		statusCode int
+		err        error
+	}{
+		{
+			name:   "successfully got offhcain data",
+			hashes: []common.Hash{common.BytesToHash([]byte("hash"))},
+			result: fmt.Sprintf(`{"result":{"%s":"%s"}}`,
+				common.BytesToHash([]byte("hash")).Hex(), hex.EncodeToString([]byte("offchaindata"))),
+			data: map[common.Hash][]byte{
+				common.BytesToHash([]byte("hash")): []byte("offchaindata"),
+			},
+		},
+		{
+			name:   "error returned by server",
+			hashes: []common.Hash{common.BytesToHash([]byte("hash"))},
+			result: `{"error":{"code":123,"message":"test error"}}`,
+			err:    errors.New("123 test error"),
+		},
+		{
+			name:   "invalid offchain data returned by server",
+			hashes: []common.Hash{common.BytesToHash([]byte("hash"))},
+			result: fmt.Sprintf(`{"result":{"%s":"invalid-signature"}}`,
+				common.BytesToHash([]byte("hash")).Hex()),
+			data: map[common.Hash][]byte{
+				common.BytesToHash([]byte("hash")): nil,
+			},
+		},
+		{
+			name:       "unsuccessful status code returned by server",
+			hashes:     []common.Hash{common.BytesToHash([]byte("hash"))},
+			statusCode: http.StatusUnauthorized,
+			err:        errors.New("invalid status code, expected: 200, found: 401"),
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			svr := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				var res rpc.Request
+				require.NoError(t, json.NewDecoder(r.Body).Decode(&res))
+				require.Equal(t, "sync_listOffChainData", res.Method)
+
+				var params [][]common.Hash
+				require.NoError(t, json.Unmarshal(res.Params, &params))
+				require.Equal(t, tt.hashes, params[0])
+
+				if tt.statusCode > 0 {
+					w.WriteHeader(tt.statusCode)
+				}
+
+				_, err := fmt.Fprint(w, tt.result)
+				require.NoError(t, err)
+			}))
+			defer svr.Close()
+
+			c := &client{url: svr.URL}
+
+			got, err := c.ListOffChainData(context.Background(), tt.hashes)
+			if tt.err != nil {
+				require.Error(t, err)
+				require.EqualError(t, tt.err, err.Error())
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, tt.data, got)
+			}
+		})
+	}
+}

cmd/main.go

@@ -142,11 +142,11 @@ func start(cliCtx *cli.Context) error {
 		[]rpc.Service{
 			{
 				Name:    sync.APISYNC,
-				Service: sync.NewSyncEndpoints(storage),
+				Service: sync.NewEndpoints(storage),
 			},
 			{
 				Name: datacom.APIDATACOM,
-				Service: datacom.NewDataComEndpoints(
+				Service: datacom.NewEndpoints(
 					storage,
 					pk,
 					sequencerTracker,

db/db.go

@@ -29,6 +29,7 @@ type DB interface {
 
 	Exists(ctx context.Context, key common.Hash) bool
 	GetOffChainData(ctx context.Context, key common.Hash, dbTx sqlx.QueryerContext) (types.ArgBytes, error)
+	ListOffChainData(ctx context.Context, keys []common.Hash, dbTx sqlx.QueryerContext) (map[common.Hash]types.ArgBytes, error)
 	StoreOffChainData(ctx context.Context, od []types.OffChainData, dbTx sqlx.ExecerContext) error
 }
 
@@ -218,6 +219,54 @@ func (db *pgDB) GetOffChainData(ctx context.Context, key common.Hash, dbTx sqlx.
 	return common.FromHex(hexValue), nil
 }
 
+// ListOffChainData returns values identified by the given keys
+func (db *pgDB) ListOffChainData(ctx context.Context, keys []common.Hash, dbTx sqlx.QueryerContext) (map[common.Hash]types.ArgBytes, error) {
+	if len(keys) == 0 {
+		return nil, nil
+	}
+
+	const listOffchainDataSQL = `
+		SELECT key, value
+		FROM data_node.offchain_data 
+		WHERE key IN (?);
+	`
+
+	preparedKeys := make([]string, len(keys))
+	for i, key := range keys {
+		preparedKeys[i] = key.Hex()
+	}
+
+	query, args, err := sqlx.In(listOffchainDataSQL, preparedKeys)
+	if err != nil {
+		return nil, err
+	}
+
+	// sqlx.In returns queries with the `?` bindvar, we can rebind it for our backend
+	query = db.pg.Rebind(query)
+
+	rows, err := db.querier(dbTx).QueryxContext(ctx, query, args...)
+	if err != nil {
+		return nil, err
+	}
+
+	defer rows.Close()
+
+	list := make(map[common.Hash]types.ArgBytes)
+	for rows.Next() {
+		data := struct {
+			Key   string `db:"key"`
+			Value string `db:"value"`
+		}{}
+		if err = rows.StructScan(&data); err != nil {
+			return nil, err
+		}
+
+		list[common.HexToHash(data.Key)] = common.FromHex(data.Value)
+	}
+
+	return list, nil
+}
+
 func (db *pgDB) execer(dbTx sqlx.ExecerContext) sqlx.ExecerContext {
 	if dbTx != nil {
 		return dbTx

db/db_test.go

@@ -2,6 +2,7 @@ package db
 
 import (
 	"context"
+	"database/sql/driver"
 	"errors"
 	"testing"
 
@@ -478,6 +479,125 @@ func Test_DB_GetOffChainData(t *testing.T) {
 	}
 }
 
+func Test_DB_ListOffChainData(t *testing.T) {
+	testTable := []struct {
+		name      string
+		od        []types.OffChainData
+		keys      []common.Hash
+		expected  map[common.Hash]types.ArgBytes
+		sql       string
+		returnErr error
+	}{
+		{
+			name: "successfully selected one value",
+			od: []types.OffChainData{{
+				Key:   common.HexToHash("key1"),
+				Value: []byte("value1"),
+			}},
+			keys: []common.Hash{
+				common.BytesToHash([]byte("key1")),
+			},
+			expected: map[common.Hash]types.ArgBytes{
+				common.BytesToHash([]byte("key1")): []byte("value1"),
+			},
+			sql: `SELECT key, value FROM data_node\.offchain_data WHERE key IN \(\$1\)`,
+		},
+		{
+			name: "successfully selected two values",
+			od: []types.OffChainData{{
+				Key:   common.HexToHash("key1"),
+				Value: []byte("value1"),
+			}, {
+				Key:   common.HexToHash("key2"),
+				Value: []byte("value2"),
+			}},
+			keys: []common.Hash{
+				common.BytesToHash([]byte("key1")),
+				common.BytesToHash([]byte("key2")),
+			},
+			expected: map[common.Hash]types.ArgBytes{
+				common.BytesToHash([]byte("key1")): []byte("value1"),
+				common.BytesToHash([]byte("key2")): []byte("value2"),
+			},
+			sql: `SELECT key, value FROM data_node\.offchain_data WHERE key IN \(\$1\, \$2\)`,
+		},
+		{
+			name: "error returned",
+			od: []types.OffChainData{{
+				Key:   common.HexToHash("key1"),
+				Value: []byte("value1"),
+			}},
+			keys: []common.Hash{
+				common.BytesToHash([]byte("key1")),
+			},
+			sql:       `SELECT key, value FROM data_node\.offchain_data WHERE key IN \(\$1\)`,
+			returnErr: errors.New("test error"),
+		},
+		{
+			name: "no rows",
+			od: []types.OffChainData{{
+				Key:   common.HexToHash("key1"),
+				Value: []byte("value1"),
+			}},
+			keys: []common.Hash{
+				common.BytesToHash([]byte("underfined")),
+			},
+			sql:       `SELECT key, value FROM data_node\.offchain_data WHERE key IN \(\$1\)`,
+			returnErr: ErrStateNotSynchronized,
+		},
+	}
+
+	for _, tt := range testTable {
+		tt := tt
+
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			db, mock, err := sqlmock.New()
+			require.NoError(t, err)
+
+			defer db.Close()
+
+			wdb := sqlx.NewDb(db, "postgres")
+
+			// Seed data
+			seedOffchainData(t, wdb, mock, tt.od)
+
+			preparedKeys := make([]driver.Value, len(tt.keys))
+			for i, key := range tt.keys {
+				preparedKeys[i] = key.Hex()
+			}
+
+			expected := mock.ExpectQuery(tt.sql).
+				WithArgs(preparedKeys...)
+
+			if tt.returnErr != nil {
+				expected.WillReturnError(tt.returnErr)
+			} else {
+				returnData := sqlmock.NewRows([]string{"key", "value"})
+
+				for key, val := range tt.expected {
+					returnData = returnData.AddRow(key.Hex(), common.Bytes2Hex(val))
+				}
+
+				expected.WillReturnRows(returnData)
+			}
+
+			dbPG := New(wdb)
+
+			data, err := dbPG.ListOffChainData(context.Background(), tt.keys, wdb)
+			if tt.returnErr != nil {
+				require.ErrorIs(t, err, tt.returnErr)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, tt.expected, data)
+			}
+
+			require.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
+}
+
 func Test_DB_Exist(t *testing.T) {
 	testTable := []struct {
 		name      string