update validation file

Signed-off-by: Musilah <[email protected]>
absmach · Feb 28, 2024 · f7d9899 · f7d9899
1 parent bd59b13
commit f7d9899
Show file tree

Hide file tree

Showing 5 changed files with 11 additions and 34 deletions.
diff --git a/internal/apiutil/errors.go b/internal/apiutil/errors.go
@@ -164,4 +164,7 @@ var (
 
 	// ErrRollbackTx indicates failed to rollback transaction.
 	ErrRollbackTx = errors.New("failed to rollback transaction")
+
+	// ErrAggregation indicates missing aggregation and interval.
+	ErrAggregation = errors.New("both aggregation and interval are required")
 )
diff --git a/pkg/sdk/go/message.go b/pkg/sdk/go/message.go
@@ -25,9 +25,9 @@ func (sdk mgSDK) SendMessage(chanName, msg, key string) errors.SDKError {
 		subtopicPart = fmt.Sprintf("/%s", strings.ReplaceAll(chanNameParts[1], ".", "/"))
 	}
 
-	reqUrl := fmt.Sprintf("%s/channels/%s/messages%s", sdk.httpAdapterURL, chanID, subtopicPart)
+	reqURL := fmt.Sprintf("%s/channels/%s/messages%s", sdk.httpAdapterURL, chanID, subtopicPart)
 
-	_, _, err := sdk.processRequest(http.MethodPost, reqUrl, ThingPrefix+key, []byte(msg), nil, http.StatusAccepted)
+	_, _, err := sdk.processRequest(http.MethodPost, reqURL, ThingPrefix+key, []byte(msg), nil, http.StatusAccepted)
 
 	return err
 }

diff --git a/readers/api/requests.go b/readers/api/requests.go
@@ -39,5 +39,9 @@ func (req listMessagesReq) validate() error {
 		return apiutil.ErrInvalidComparator
 	}
 
+	if req.pageMeta.Aggregation == "" || req.pageMeta.Interval == "" {
+		return apiutil.ErrAggregation
+	}
+
 	return nil
 }
diff --git a/readers/api/transport.go b/readers/api/transport.go
@@ -153,10 +153,6 @@ func decodeList(_ context.Context, r *http.Request) (interface{}, error) {
 		return nil, errors.Wrap(apiutil.ErrValidation, err)
 	}
 
-	if aggregation == "" || interval == "" {
-		return nil, errors.Wrap(apiutil.ErrValidation, errors.New("both aggregation and interval are required"))
-	}
-
 	req := listMessagesReq{
 		chanID: chi.URLParam(r, "chanID"),
 		token:  apiutil.ExtractBearerToken(r),

diff --git a/readers/timescale/messages.go b/readers/timescale/messages.go
@@ -42,22 +42,7 @@ func (tr timescaleRepository) ReadAll(chanID string, rpm readers.PageMetadata) (
 	// If aggregation is provided, add time_bucket and aggregation to the query
 	switch {
 	case rpm.Aggregation != "":
-		q = fmt.Sprintf(`
-			SELECT 
-			EXTRACT(epoch FROM time_bucket('%s', to_timestamp(time))) AS time, 
-			%s(value) AS value 
-			%s
-			ORDER BY
-			%s DESC
-			LIMIT 
-			:limit 
-			OFFSET 
-			:offset;`,
-			rpm.Interval,
-			rpm.Aggregation,
-			baseQuery,
-			order,
-		)
+		q = fmt.Sprintf(`SELECT EXTRACT(epoch FROM time_bucket('%s', to_timestamp(time))) AS time, %s(value) AS value %s ORDER BY %s DESC LIMIT :limit OFFSET :offset;`, rpm.Interval, rpm.Aggregation, baseQuery, order)
 	default:
 		// Construct the base query without time_bucket and aggregation
 		q = fmt.Sprintf(`SELECT * FROM %s WHERE %s ORDER BY %s DESC LIMIT :limit OFFSET :offset;`, format, fmtCondition(chanID, rpm), order)
@@ -121,18 +106,7 @@ func (tr timescaleRepository) ReadAll(chanID string, rpm readers.PageMetadata) (
 	var countQuery string
 	switch {
 	case rpm.Aggregation != "":
-		countQuery = fmt.Sprintf(`
-			SELECT COUNT(*) 
-			FROM (
-				SELECT 
-				EXTRACT(epoch FROM time_bucket('%s', to_timestamp(time))) AS time, 
-				%s(value) AS value 
-				%s
-			) AS subquery;`,
-			rpm.Interval,
-			rpm.Aggregation,
-			baseQuery,
-		)
+		countQuery = fmt.Sprintf(`SELECT COUNT(*) FROM (SELECT EXTRACT(epoch FROM time_bucket('%s', to_timestamp(time))) AS time, %s(value) AS value %s) AS subquery;`, rpm.Interval, rpm.Aggregation, baseQuery)
 	default:
 		countQuery = fmt.Sprintf(`SELECT COUNT(*) FROM %s WHERE %s;`, format, fmtCondition(chanID, rpm))
 	}