add api reset price (#118)

* add api set null price

* add from-to

* add condition for time range

* split logic and update readme

* add comment

* .

README.md

@@ -259,4 +259,44 @@ Example: generate `Storage` interface to a struct name `MockStorage`
 mockery --dir=v2/pkg/storage/state --name=Storage --output=v2/mocks/ --structname=MockState --filename=State.go
 ```
 
-For more information `mockery --help`
+For more information `mockery --help`
+
+## Price-filler
+### Refetch token price 
+
+- **URL**: `/price_filler/refetch`
+- **Method**: `POST`
+- **Description**: Reset taker_token_price, maker_token_price, taker_usd_amount, maker_usd_amount of trades with faulty token.
+- **URL Parameters**:
+    - `address` (string): The token address.
+    - `exchange` (string): The exchange name.
+    - `from_ts` (int): The starting timestamp (millisecond).
+    - `to_ts` (int): The ending timestamp (millisecond).
+- **Response**:
+    - **200 OK**: Success.
+      ```json
+      {
+        "data": {
+            "token": "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
+            "exchange": "bebop",
+            "from": 1734204504800,
+            "to": 1734809304800,
+            "number of row updated": 2000
+        },
+        "success": true
+      }
+      ```
+    - **400 Bad Request**: If the from_ts or to_ts is invalid.
+      ```json
+      {
+          "error": "<error_message>",
+          "success": false
+      }
+      ```
+    - **500 Internal Server Error**: If there is an error resetting token price.
+      ```json
+      {
+        "error": "<error_message>",
+        "success": false
+      }
+      ```

v2/internal/server/tradelogs.go

@@ -17,6 +17,19 @@ var (
 	maxTimeRange = uint64(24 * time.Hour.Milliseconds())
 )
 
+const (
+	maxRangeEmptyAddress    = 7 * 24 * time.Hour
+	maxRangeNonEmptyAddress = 30 * 24 * time.Hour
+)
+
+type resetTokenPriceParams struct {
+	Address  string `form:"address" json:"address"`
+	Exchange string `form:"exchange" json:"exchange"`
+	From     int64  `form:"from_ts" json:"from_ts"`
+	To       int64  `form:"to_ts" json:"to_ts"`
+	Rows     int64  `form:"rows" json:"rows"`
+}
+
 type TradeLogs struct {
 	r           *gin.Engine
 	bindAddr    string
@@ -65,6 +78,7 @@ func (s *TradeLogs) register() {
 	s.r.POST("/makers", s.addMakerName)
 	s.r.GET("/txorigin", s.getTxOrigin)
 	s.r.POST("/txorigin", s.addTxOrigin)
+	s.r.POST("/price_filler/refetch", s.resetTokenPriceToRefetch)
 }
 
 func (s *TradeLogs) getTradeLogs(c *gin.Context) {
@@ -190,3 +204,74 @@ func (s *TradeLogs) addTxOrigin(c *gin.Context) {
 		"data":    queries,
 	})
 }
+
+func (s *TradeLogs) resetTokenPriceToRefetch(c *gin.Context) {
+	var query resetTokenPriceParams
+	if err := c.ShouldBindJSON(&query); err != nil {
+		responseErr(c, http.StatusBadRequest, err)
+		return
+	}
+
+	query, err := validateResetTokenPriceParams(query)
+	if err != nil {
+		responseErr(c, http.StatusBadRequest, err)
+		return
+	}
+
+	for _, storage := range s.storage {
+		if storage.Exchange() != query.Exchange {
+			continue
+		}
+		rows, err := storage.ResetTokenPriceToRefetch(query.Address, query.From, query.To)
+		if err != nil {
+			responseErr(c, http.StatusInternalServerError, err)
+			return
+		}
+		query.Rows = rows
+		c.JSON(http.StatusOK, gin.H{
+			"success": true,
+			"data":    query,
+		})
+		return
+	}
+	responseErr(c, http.StatusBadRequest, fmt.Errorf("exchange not found"))
+}
+
+func validateResetTokenPriceParams(query resetTokenPriceParams) (resetTokenPriceParams, error) {
+	now := time.Now()
+	if query.Address == "" && query.From == 0 && query.To == 0 {
+		return query, fmt.Errorf("address is empty and no valid time range provided")
+	}
+
+	// Validate `From` timestamp: it cannot be in the future or negative.
+	if query.From > now.UnixMilli() || query.From < 0 {
+		return query, fmt.Errorf("invalid from_ts")
+	}
+
+	// Validate `To` timestamp: it can not be negative,
+	// and if provided, it must be greater than or equal to `From`.
+	if query.To < 0 || (query.To > 0 && query.To < query.From) {
+		return query, fmt.Errorf("invalid to_ts")
+	}
+	timeRange := maxRangeEmptyAddress
+	if len(query.Address) > 0 {
+		timeRange = maxRangeNonEmptyAddress
+	}
+
+	if query.From == 0 && query.To == 0 {
+		query.From = now.Add(-timeRange).UnixMilli()
+		query.To = now.UnixMilli()
+		return query, nil
+	}
+
+	// If `From` is provided but `To` is not, calculate `To` as `From + max range`
+	if query.From > 0 && query.To == 0 {
+		query.To = min(time.UnixMilli(query.From).Add(timeRange).UnixMilli(), now.UnixMilli())
+		return query, nil
+	}
+
+	// Adjust `From` as `To - max range
+	query.To = min(query.To, now.UnixMilli())
+	query.From = max(time.UnixMilli(query.To).Add(-timeRange).UnixMilli(), query.From)
+	return query, nil
+}

v2/pkg/storage/tradelogs/bebop/storage.go

@@ -212,3 +212,36 @@ func tradeLogColumns() []string {
 		"expiration",
 	}
 }
+
+func (s *Storage) ResetTokenPriceToRefetch(token string, from, to int64) (int64, error) {
+	builder := squirrel.StatementBuilder.PlaceholderFormat(squirrel.Dollar).
+		Update(s.tableName()).
+		Set("maker_token_price", nil).
+		Set("taker_token_price", nil).
+		Set("maker_usd_amount", nil).
+		Set("taker_usd_amount", nil)
+	if token != "" {
+		builder = builder.Where(squirrel.Or{
+			squirrel.Eq{"maker_token": token},
+			squirrel.Eq{"taker_token": token},
+		})
+	}
+	builder = builder.Where(squirrel.And{
+		squirrel.GtOrEq{"timestamp": from},
+		squirrel.LtOrEq{"timestamp": to},
+	})
+	q, p, err := builder.ToSql()
+
+	if err != nil {
+		return 0, fmt.Errorf("build query error: %w", err)
+	}
+	result, err := s.db.Exec(q, p...)
+	if err != nil {
+		return 0, fmt.Errorf("run query error: %w", err)
+	}
+	rowsAffected, err := result.RowsAffected()
+	if err != nil {
+		return 0, fmt.Errorf("fetch rows affected error: %w", err)
+	}
+	return rowsAffected, nil
+}

v2/pkg/storage/tradelogs/hashflow_v3/storage.go

@@ -128,3 +128,36 @@ func (s *Storage) Delete(blocks []uint64) error {
 	}
 	return nil
 }
+
+func (s *Storage) ResetTokenPriceToRefetch(token string, from, to int64) (int64, error) {
+	builder := squirrel.StatementBuilder.PlaceholderFormat(squirrel.Dollar).
+		Update(s.tableName()).
+		Set("maker_token_price", nil).
+		Set("taker_token_price", nil).
+		Set("maker_usd_amount", nil).
+		Set("taker_usd_amount", nil)
+	if token != "" {
+		builder = builder.Where(squirrel.Or{
+			squirrel.Eq{"maker_token": token},
+			squirrel.Eq{"taker_token": token},
+		})
+	}
+	builder = builder.Where(squirrel.And{
+		squirrel.GtOrEq{"timestamp": from},
+		squirrel.LtOrEq{"timestamp": to},
+	})
+	q, p, err := builder.ToSql()
+
+	if err != nil {
+		return 0, fmt.Errorf("build query error: %w", err)
+	}
+	result, err := s.db.Exec(q, p...)
+	if err != nil {
+		return 0, fmt.Errorf("run query error: %w", err)
+	}
+	rowsAffected, err := result.RowsAffected()
+	if err != nil {
+		return 0, fmt.Errorf("fetch rows affected error: %w", err)
+	}
+	return rowsAffected, nil
+}

v2/pkg/storage/tradelogs/kyberswap/storage.go

@@ -128,3 +128,36 @@ func (s *Storage) Delete(blocks []uint64) error {
 	}
 	return nil
 }
+
+func (s *Storage) ResetTokenPriceToRefetch(token string, from, to int64) (int64, error) {
+	builder := squirrel.StatementBuilder.PlaceholderFormat(squirrel.Dollar).
+		Update(s.tableName()).
+		Set("maker_token_price", nil).
+		Set("taker_token_price", nil).
+		Set("maker_usd_amount", nil).
+		Set("taker_usd_amount", nil)
+	if token != "" {
+		builder = builder.Where(squirrel.Or{
+			squirrel.Eq{"maker_token": token},
+			squirrel.Eq{"taker_token": token},
+		})
+	}
+	builder = builder.Where(squirrel.And{
+		squirrel.GtOrEq{"timestamp": from},
+		squirrel.LtOrEq{"timestamp": to},
+	})
+	q, p, err := builder.ToSql()
+
+	if err != nil {
+		return 0, fmt.Errorf("build query error: %w", err)
+	}
+	result, err := s.db.Exec(q, p...)
+	if err != nil {
+		return 0, fmt.Errorf("run query error: %w", err)
+	}
+	rowsAffected, err := result.RowsAffected()
+	if err != nil {
+		return 0, fmt.Errorf("fetch rows affected error: %w", err)
+	}
+	return rowsAffected, nil
+}

v2/pkg/storage/tradelogs/kyberswap_rfq/storage.go

@@ -128,3 +128,36 @@ func (s *Storage) Delete(blocks []uint64) error {
 	}
 	return nil
 }
+
+func (s *Storage) ResetTokenPriceToRefetch(token string, from, to int64) (int64, error) {
+	builder := squirrel.StatementBuilder.PlaceholderFormat(squirrel.Dollar).
+		Update(s.tableName()).
+		Set("maker_token_price", nil).
+		Set("taker_token_price", nil).
+		Set("maker_usd_amount", nil).
+		Set("taker_usd_amount", nil)
+	if token != "" {
+		builder = builder.Where(squirrel.Or{
+			squirrel.Eq{"maker_token": token},
+			squirrel.Eq{"taker_token": token},
+		})
+	}
+	builder = builder.Where(squirrel.And{
+		squirrel.GtOrEq{"timestamp": from},
+		squirrel.LtOrEq{"timestamp": to},
+	})
+	q, p, err := builder.ToSql()
+
+	if err != nil {
+		return 0, fmt.Errorf("build query error: %w", err)
+	}
+	result, err := s.db.Exec(q, p...)
+	if err != nil {
+		return 0, fmt.Errorf("run query error: %w", err)
+	}
+	rowsAffected, err := result.RowsAffected()
+	if err != nil {
+		return 0, fmt.Errorf("fetch rows affected error: %w", err)
+	}
+	return rowsAffected, nil
+}

v2/pkg/storage/tradelogs/oneinch_v6/storage.go

@@ -218,3 +218,36 @@ func tradeLogColumns() []string {
 		"expiration",
 	}
 }
+
+func (s *Storage) ResetTokenPriceToRefetch(token string, from, to int64) (int64, error) {
+	builder := squirrel.StatementBuilder.PlaceholderFormat(squirrel.Dollar).
+		Update(s.tableName()).
+		Set("maker_token_price", nil).
+		Set("taker_token_price", nil).
+		Set("maker_usd_amount", nil).
+		Set("taker_usd_amount", nil)
+	if token != "" {
+		builder = builder.Where(squirrel.Or{
+			squirrel.Eq{"maker_token": token},
+			squirrel.Eq{"taker_token": token},
+		})
+	}
+	builder = builder.Where(squirrel.And{
+		squirrel.GtOrEq{"timestamp": from},
+		squirrel.LtOrEq{"timestamp": to},
+	})
+	q, p, err := builder.ToSql()
+
+	if err != nil {
+		return 0, fmt.Errorf("build query error: %w", err)
+	}
+	result, err := s.db.Exec(q, p...)
+	if err != nil {
+		return 0, fmt.Errorf("run query error: %w", err)
+	}
+	rowsAffected, err := result.RowsAffected()
+	if err != nil {
+		return 0, fmt.Errorf("fetch rows affected error: %w", err)
+	}
+	return rowsAffected, nil
+}