feat: enable aggregates and group_by

Author: catalyst17

api/api.go

@@ -30,7 +30,7 @@ type QueryParams struct {
 	// @Description Map of filter parameters
 	FilterParams map[string]string `schema:"-"`
 	// @Description Field to group results by
-	GroupBy string `schema:"group_by"`
+	GroupBy []string `schema:"group_by"`
 	// @Description Field to sort results by
 	SortBy string `schema:"sort_by"`
 	// @Description Sort order (asc or desc)

internal/handlers/logs_handlers.go

@@ -135,7 +135,7 @@ func handleLogsRequest(c *gin.Context, contractAddress, signature string) {
 
 	logs, err := mainStorage.GetLogs(storage.QueryFilter{
 		FilterParams:    queryParams.FilterParams,
-		GroupBy:         []string{queryParams.GroupBy},
+		GroupBy:         queryParams.GroupBy,
 		SortBy:          queryParams.SortBy,
 		SortOrder:       queryParams.SortOrder,
 		Page:            queryParams.Page,

internal/handlers/transactions_handlers.go

@@ -136,7 +136,7 @@ func handleTransactionsRequest(c *gin.Context, contractAddress, signature string
 
 	result, err := mainStorage.GetTransactions(storage.QueryFilter{
 		FilterParams:    queryParams.FilterParams,
-		GroupBy:         []string{queryParams.GroupBy},
+		GroupBy:         queryParams.GroupBy,
 		SortBy:          queryParams.SortBy,
 		SortOrder:       queryParams.SortOrder,
 		Page:            queryParams.Page,

internal/storage/clickhouse.go

@@ -342,9 +342,19 @@ func (c *ClickHouseConnector) GetTransactions(qf QueryFilter) (QueryResult[commo
 	return executeQuery[common.Transaction](c, "transactions", columns, qf, scanTransaction)
 }
 
-func (c *ClickHouseConnector) GetLogs(qf QueryFilter) (QueryResult[common.Log], 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"
-	return executeQuery[common.Log](c, "logs", columns, qf, scanLog)
+func (c *ClickHouseConnector) GetLogs(qf QueryFilter) (QueryResult[map[string]interface{}], error) {
+	var columns string
+
+	if len(qf.GroupBy) > 0 || len(qf.Aggregates) > 0 {
+		// Build columns for SELECT when grouping or aggregating
+		selectColumns := append(qf.GroupBy, qf.Aggregates...)
+		columns = strings.Join(selectColumns, ", ")
+	} else {
+		// Default columns when not grouping
+		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"
+	}
+
+	return executeQuery[map[string]interface{}](c, "logs", columns, qf, scanRowToMap)
 }
 
 func executeQuery[T any](c *ClickHouseConnector, table, columns string, qf QueryFilter, scanFunc func(driver.Rows) (T, error)) (QueryResult[T], error) {
@@ -397,7 +407,13 @@ func (c *ClickHouseConnector) buildQuery(table, columns string, qf QueryFilter)
 		query = addFilterParams(key, strings.ToLower(value), query)
 	}
 
-	// Add sort by clause
+	// Add GROUP BY clause if specified
+	if len(qf.GroupBy) > 0 {
+		groupByColumns := strings.Join(qf.GroupBy, ", ")
+		query += fmt.Sprintf(" GROUP BY %s", groupByColumns)
+	}
+
+	// Add ORDER BY clause
 	if qf.SortBy != "" {
 		query += fmt.Sprintf(" ORDER BY %s %s", qf.SortBy, qf.SortOrder)
 	}
@@ -556,6 +572,27 @@ func scanLog(rows driver.Rows) (common.Log, error) {
 	return log, nil
 }
 
+func scanRowToMap(rows driver.Rows) (map[string]interface{}, error) {
+	columns := rows.Columns()
+	values := make([]interface{}, len(columns))
+	valuePtrs := make([]interface{}, len(columns))
+
+	for i := range columns {
+		valuePtrs[i] = &values[i]
+	}
+
+	if err := rows.Scan(valuePtrs...); err != nil {
+		return nil, err
+	}
+
+	result := make(map[string]interface{})
+	for i, col := range columns {
+		result[col] = values[i]
+	}
+
+	return result, nil
+}
+
 func (c *ClickHouseConnector) GetMaxBlockNumber(chainId *big.Int) (maxBlockNumber *big.Int, err error) {
 	query := fmt.Sprintf("SELECT number FROM %s.blocks WHERE is_deleted = 0", c.cfg.Database)
 	if chainId.Sign() > 0 {

internal/storage/connector.go

@@ -54,7 +54,7 @@ type IMainStorage interface {
 
 	GetBlocks(qf QueryFilter) (blocks []common.Block, err error)
 	GetTransactions(qf QueryFilter) (transactions QueryResult[common.Transaction], err error)
-	GetLogs(qf QueryFilter) (logs QueryResult[common.Log], err error)
+	GetLogs(qf QueryFilter) (logs QueryResult[map[string]interface{}], err error)
 	GetTraces(qf QueryFilter) (traces []common.Trace, err error)
 	GetMaxBlockNumber(chainId *big.Int) (maxBlockNumber *big.Int, err error)
 	/**