Implement GetLogs queries without grouping and filter columns (#44)

### TL;DR

Updated API response structure, improved query filtering, and added aggregation support for logs.

### What changed?

- Modified `QueryResponse` structure in `api.go` to use `interface{}` for `Data` field and `map[string]string` for `Aggregations`.
- Updated `ParseQueryParams` function to handle `event_name` filter.
- Changed `GroupBy` parameter in `QueryFilter` to accept an array of strings.
- Refactored `GetLogs` function in `clickhouse.go` to support advanced filtering, sorting, and aggregations.
- Updated `IMainStorage` interface to return `QueryResult` instead of `[]common.Log` for `GetLogs`.
- Added debug binary files to `.gitignore`.

### How to test?

1. Run the API and make requests to the logs endpoint with various query parameters.
2. Test filtering using `filter_event_name` and other filter parameters.
3. Verify that grouping, sorting, and pagination work as expected.
4. Check if aggregations are correctly calculated and returned in the response.

### Why make this change?

These changes improve the flexibility and functionality of the API, allowing for more complex queries and data analysis. The updated structure supports better filtering, aggregation, and response formatting, which enhances the overall usability of the API for clients.

Author: AmineAfia

.gitignore

@@ -24,7 +24,9 @@ go.work.sum
 # env file
 .env
 
+# dev files
+cmd/api/__debug_bin*
 .vscode
 
 configs/secrets*
-!configs/secrets.example.yml
+!configs/secrets.example.yml

api/api.go

@@ -39,9 +39,9 @@ type Meta struct {
 }
 
 type QueryResponse struct {
-	Meta Meta          `json:"meta"`
-	Data []interface{} `json:"data"`
-	Aggregations map[string]interface{} `json:"aggregations,omitempty"`
+	Meta         Meta                   `json:"meta"`
+	Data         interface{}            `json:"data,omitempty"`
+	Aggregations map[string]string      `json:"aggregations,omitempty"`
 }
 
 func writeError(w http.ResponseWriter, message string, code int) {
@@ -75,7 +75,12 @@ func ParseQueryParams(r *http.Request) (QueryParams, error) {
 	params.FilterParams = make(map[string]string)
 	for key, values := range rawQueryParams {
 		if strings.HasPrefix(key, "filter_") {
-			params.FilterParams[key] = values[0]
+			// TODO: tmp hack remove it once we implement filtering with operators
+			strippedKey := strings.Replace(key, "filter_", "", 1)
+			if strippedKey == "event_name" {
+				strippedKey = "data"
+			}
+			params.FilterParams[strippedKey] = values[0]
 			delete(rawQueryParams, key)
 		}
 	}

internal/handlers/logs_handlers.go

@@ -56,7 +56,7 @@ func handleLogsRequest(w http.ResponseWriter, r *http.Request, contractAddress,
 
 	logs, err := mainStorage.GetLogs(storage.QueryFilter{
 		FilterParams:    queryParams.FilterParams,
-		GroupBy:         queryParams.GroupBy,
+		GroupBy:         []string{queryParams.GroupBy},
 		SortBy:          queryParams.SortBy,
 		SortOrder:       queryParams.SortOrder,
 		Page:            queryParams.Page,
@@ -81,10 +81,8 @@ func handleLogsRequest(w http.ResponseWriter, r *http.Request, contractAddress,
 			TotalItems:      0, // TODO: Implement total items count
 			TotalPages:      0, // TODO: Implement total pages count
 		},
-		Data: []interface{}{logs},
-		Aggregations: map[string]interface{}{
-			"aggregates": queryParams.Aggregates,
-		},
+		Data:         logs.Data,
+		Aggregations: logs.Aggregates,
 	}
 
 	sendJSONResponse(w, response)

internal/handlers/transactions_handlers.go

@@ -46,7 +46,7 @@ func handleTransactionsRequest(w http.ResponseWriter, r *http.Request, contractA
 
 	transactions, err := mainStorage.GetTransactions(storage.QueryFilter{
 		FilterParams:    queryParams.FilterParams,
-		GroupBy:         queryParams.GroupBy,
+		GroupBy:         []string{queryParams.GroupBy},
 		SortBy:          queryParams.SortBy,
 		SortOrder:       queryParams.SortOrder,
 		Page:            queryParams.Page,
@@ -71,10 +71,8 @@ func handleTransactionsRequest(w http.ResponseWriter, r *http.Request, contractA
 			TotalItems:      0, // TODO: Implement total items count
 			TotalPages:      0, // TODO: Implement total pages count
 		},
-		Data: []interface{}{transactions},
-		Aggregations: map[string]interface{}{
-			"aggregates": queryParams.Aggregates,
-		},
+		Data: transactions,
+		// Aggregations: queryParams.Aggregates,
 	}
 
 	sendJSONResponse(w, response)

internal/storage/clickhouse.go

@@ -312,17 +312,53 @@ func (c *ClickHouseConnector) GetTransactions(qf QueryFilter) (txs []common.Tran
 	return txs, nil
 }
 
-func (c *ClickHouseConnector) GetLogs(qf QueryFilter) (logs []common.Log, err error) {
+func (c *ClickHouseConnector) GetLogs(qf QueryFilter) (QueryResult, error) {
 	columns := "chain_id, block_number, block_hash, block_timestamp, transaction_hash, transaction_index, log_index, address, data, topic_0, topic_1, topic_2, topic_3"
-	query := fmt.Sprintf("SELECT %s FROM %s.logs FINAL WHERE is_deleted = 0%s",
-		columns, c.cfg.Database, getLimitClause(int(qf.Limit)))
-	// TODO: add aggregations, filters and grouping
+	
+	// Start building the query
+	query := fmt.Sprintf("SELECT %s FROM %s.logs FINAL WHERE is_deleted = 0", columns, c.cfg.Database)
+
+	// Add filters
+	if qf.ContractAddress != "" {
+		query += fmt.Sprintf(" AND address = '%s'", qf.ContractAddress)
+	}
+	if qf.FunctionSig != "" {
+		query += fmt.Sprintf(" AND topic_0 = '%s'", qf.FunctionSig)
+	}
+	if len(qf.FilterParams) > 0 {
+		for key, value := range qf.FilterParams {
+			query += fmt.Sprintf(" AND %s = '%s'", key, value)
+		}
+	}
+
+	// Add grouping (TODO: fix it by adding all needed columns)
+	// if len(qf.GroupBy) > 0 {
+	// 	query += " GROUP BY " + strings.Join(qf.GroupBy, ", ")
+	// }
+
+	// Add sorting
+	if qf.SortBy != "" {
+		query += fmt.Sprintf(" ORDER BY %s %s", qf.SortBy, qf.SortOrder)
+	}
+
+	// Add pagination
+	if qf.Page > 0 && qf.Limit > 0 {
+		offset := (qf.Page - 1) * qf.Limit
+		query += fmt.Sprintf(" LIMIT %d OFFSET %d", qf.Limit, offset)
+	} else {
+		query += getLimitClause(int(qf.Limit))
+	}
+
+	// Execute the query
 	rows, err := c.conn.Query(context.Background(), query)
 	if err != nil {
-		return nil, err
+		return QueryResult{Data: []common.Log{}, Aggregates: map[string]string{}}, err
 	}
 	defer rows.Close()
-
+	queryResult := QueryResult{
+		Data: []common.Log{},
+		Aggregates: map[string]string{},
+	}
 	for rows.Next() {
 		var log common.Log
 		var timestamp uint64
@@ -344,17 +380,61 @@ func (c *ClickHouseConnector) GetLogs(qf QueryFilter) (logs []common.Log, err er
 		)
 		if err != nil {
 			zLog.Error().Err(err).Msg("Error scanning log")
-			return nil, err
+			return QueryResult{Data: []common.Log{}}, err
 		}
 		log.BlockTimestamp = time.Unix(int64(timestamp), 0)
 		for _, topic := range topics {
 			if topic != "" {
 				log.Topics = append(log.Topics, topic)
 			}
 		}
-		logs = append(logs, log)
+		queryResult.Data = append(queryResult.Data, log)
 	}
-	return logs, nil
+
+	// Handle aggregations
+	if len(qf.Aggregates) > 0 {
+		aggregateQuery := "SELECT "
+		for i, agg := range qf.Aggregates {
+			if i > 0 {
+				aggregateQuery += ", "
+			}
+			aggregateQuery += agg
+		}
+		aggregateQuery += " FROM " + c.cfg.Database + ".logs FINAL WHERE is_deleted = 0"
+		
+		// Add the same filters as the main query
+		if qf.ContractAddress != "" {
+			aggregateQuery += fmt.Sprintf(" AND address = '%s'", qf.ContractAddress)
+		}
+		if qf.FunctionSig != "" {
+			aggregateQuery += fmt.Sprintf(" AND topic_0 = '%s'", qf.FunctionSig)
+		}
+		if len(qf.FilterParams) > 0 {
+			for key, value := range qf.FilterParams {
+				// TODO: add parsing for filters to get the column name and operator
+				aggregateQuery += fmt.Sprintf(" AND %s = '%s'", key, value)
+			}
+		}
+
+		row := c.conn.QueryRow(context.Background(), aggregateQuery)
+		// Marshal the result to JSON
+		// TODO: needs a typing scan
+		aggregateResultsJSON, err := json.Marshal(row)
+		if err != nil {
+			zLog.Error().Err(err).Msg("Error marshaling aggregate results to JSON")
+			return queryResult, err
+		}
+
+		var aggregateResultsMap map[string]string
+		err = json.Unmarshal(aggregateResultsJSON, &aggregateResultsMap)
+		if err != nil {
+			zLog.Error().Err(err).Msg("Error unmarshaling aggregate results JSON to map")
+			return queryResult, err
+		}
+		queryResult.Aggregates = aggregateResultsMap
+	}
+
+	return queryResult, nil
 }
 
 func (c *ClickHouseConnector) GetMaxBlockNumber() (maxBlockNumber *big.Int, err error) {

internal/storage/connector.go

@@ -11,7 +11,7 @@ import (
 type QueryFilter struct {
 	BlockNumbers    []*big.Int
 	FilterParams    map[string]string
-	GroupBy         string
+	GroupBy         []string
 	SortBy          string
 	SortOrder       string
 	Page            int
@@ -20,7 +20,10 @@ type QueryFilter struct {
 	ContractAddress string
 	FunctionSig     string
 }
-
+type QueryResult struct {
+	Data       []common.Log      `json:"data"`
+	Aggregates map[string]string `json:"aggregates"`
+}
 type IStorage struct {
 	OrchestratorStorage IOrchestratorStorage
 	MainStorage         IMainStorage
@@ -50,7 +53,7 @@ type IMainStorage interface {
 
 	GetBlocks(qf QueryFilter) (logs []common.Block, err error)
 	GetTransactions(qf QueryFilter) (logs []common.Transaction, err error)
-	GetLogs(qf QueryFilter) (logs []common.Log, err error)
+	GetLogs(qf QueryFilter) (logs QueryResult, err error)
 	GetTraces(qf QueryFilter) (traces []common.Trace, err error)
 	GetMaxBlockNumber() (maxBlockNumber *big.Int, err error)
 }