NOISSUE - Add task metadata

examples/http-server/src/lib.rs

@@ -116,7 +116,7 @@ fn headers_response(request: IncomingRequest, response_out: ResponseOutparam) {
 
     let out_headers = Fields::new();
     out_headers
-        .set(&"content-type".to_string(), &[b"text/plain".to_vec()])
+        .set("content-type", &[b"text/plain".to_vec()])
         .unwrap();
     let response = OutgoingResponse::new(out_headers);
     response.set_status_code(200).unwrap();
@@ -152,9 +152,8 @@ fn read_body(request: IncomingRequest) -> Vec<u8> {
             Ok(chunk) => data.extend_from_slice(&chunk),
             Err(_) => break,
         }
-        match stream.read(0) {
-            Err(_) => break,
-            Ok(_) => {}
+        if stream.read(0).is_err() {
+            break;
         }
     }
     data

manager/api/endpoint.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/absmach/propeller/manager"
 	pkgerrors "github.com/absmach/propeller/pkg/errors"
+	"github.com/absmach/propeller/pkg/sdk"
 	apiutil "github.com/absmach/supermq/api/http/util"
 	"github.com/go-kit/kit/endpoint"
 )
@@ -224,22 +225,35 @@ func stopJobEndpoint(svc manager.Service) endpoint.Endpoint {
 
 func listTasksEndpoint(svc manager.Service) endpoint.Endpoint {
 	return func(ctx context.Context, request any) (any, error) {
-		req, ok := request.(listEntityReq)
-		if !ok {
+		switch req := request.(type) {
+		case listEntityReq:
+			if err := req.validate(); err != nil {
+				return listTaskResponse{}, errors.Join(apiutil.ErrValidation, err)
+			}
+			tasks, err := svc.ListTasks(ctx, req.offset, req.limit)
+			if err != nil {
+				return listTaskResponse{}, err
+			}
+
+			return listTaskResponse{TaskPage: tasks}, nil
+		case listTasksReq:
+			if err := req.validate(); err != nil {
+				return listTaskResponse{}, errors.Join(apiutil.ErrValidation, err)
+			}
+			pm := sdk.PageMetadata{
+				Offset:         req.offset,
+				Limit:          req.limit,
+				MetadataFilter: req.metadataFilter,
+			}
+			tasks, err := svc.ListTasksByFilter(ctx, pm)
+			if err != nil {
+				return listTaskResponse{}, err
+			}
+
+			return listTaskResponse{TaskPage: tasks}, nil
+		default:
 			return listTaskResponse{}, errors.Join(apiutil.ErrValidation, pkgerrors.ErrInvalidData)
 		}
-		if err := req.validate(); err != nil {
-			return listTaskResponse{}, errors.Join(apiutil.ErrValidation, err)
-		}
-
-		tasks, err := svc.ListTasks(ctx, req.offset, req.limit)
-		if err != nil {
-			return listTaskResponse{}, err
-		}
-
-		return listTaskResponse{
-			TaskPage: tasks,
-		}, nil
 	}
 }
 

manager/api/requests.go

@@ -1,13 +1,17 @@
 package api
 
 import (
+	"encoding/json"
+	"errors"
 	"fmt"
 
 	"github.com/absmach/propeller/pkg/cron"
 	"github.com/absmach/propeller/pkg/task"
 	apiutil "github.com/absmach/supermq/api/http/util"
 )
 
+const maxMetadataBytes = 65536
+
 type taskReq struct {
 	task.Task `json:",inline"`
 }
@@ -31,6 +35,16 @@ func (t *taskReq) validate() error {
 		return fmt.Errorf("priority must be between 0 and 100, got %d", t.Priority)
 	}
 
+	if len(t.Metadata) > 0 {
+		b, err := json.Marshal(t.Metadata)
+		if err != nil {
+			return fmt.Errorf("invalid metadata: %w", err)
+		}
+		if len(b) > maxMetadataBytes {
+			return errors.New("metadata exceeds 64KB limit")
+		}
+	}
+
 	return nil
 }
 
@@ -100,6 +114,16 @@ func (e *listEntityReq) validate() error {
 	return nil
 }
 
+type listTasksReq struct {
+	offset         uint64
+	limit          uint64
+	metadataFilter map[string]string
+}
+
+func (r *listTasksReq) validate() error {
+	return nil
+}
+
 type metricsReq struct {
 	id            string
 	offset, limit uint64

manager/api/transport.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"log/slog"
 	"net/http"
+	"regexp"
 	"strings"
 
 	"github.com/absmach/propeller/manager"
@@ -69,7 +70,7 @@ func MakeHandler(svc manager.Service, logger *slog.Logger, instanceID string) ht
 		), "create-task").ServeHTTP)
 		r.Get("/", otelhttp.NewHandler(kithttp.NewServer(
 			listTasksEndpoint(svc),
-			decodeListEntityReq,
+			decodeListTasksReq,
 			api.EncodeResponse,
 			opts...,
 		), "list-tasks").ServeHTTP)
@@ -312,6 +313,42 @@ func decodeListEntityReq(_ context.Context, r *http.Request) (any, error) {
 	}, nil
 }
 
+var metadataKeyRe = regexp.MustCompile(`^[a-zA-Z0-9._/\-]+$`)
+
+func decodeListTasksReq(_ context.Context, r *http.Request) (any, error) {
+	o, err := apiutil.ReadNumQuery[uint64](r, api.OffsetKey, api.DefOffset)
+	if err != nil {
+		return nil, errors.Join(apiutil.ErrValidation, err)
+	}
+
+	l, err := apiutil.ReadNumQuery[uint64](r, api.LimitKey, api.DefLimit)
+	if err != nil {
+		return nil, errors.Join(apiutil.ErrValidation, err)
+	}
+
+	filter := map[string]string{}
+	for key, vals := range r.URL.Query() {
+		if !strings.HasPrefix(key, "metadata[") || !strings.HasSuffix(key, "]") {
+			continue
+		}
+		metaKey := key[len("metadata[") : len(key)-1]
+		if !metadataKeyRe.MatchString(metaKey) {
+			return nil, errors.Join(apiutil.ErrValidation, errors.New("metadata key contains invalid characters"))
+		}
+		filter[metaKey] = vals[0]
+	}
+
+	if len(filter) == 0 {
+		return listEntityReq{offset: o, limit: l}, nil
+	}
+
+	return listTasksReq{
+		offset:         o,
+		limit:          l,
+		metadataFilter: filter,
+	}, nil
+}
+
 func decodeMetricsReq(key string) kithttp.DecodeRequestFunc {
 	return func(_ context.Context, r *http.Request) (any, error) {
 		o, err := apiutil.ReadNumQuery[uint64](r, api.OffsetKey, api.DefOffset)

manager/manager.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/absmach/propeller/pkg/proplet"
+	"github.com/absmach/propeller/pkg/sdk"
 	"github.com/absmach/propeller/pkg/task"
 )
 
@@ -22,6 +23,7 @@ type Service interface {
 	StartJob(ctx context.Context, jobID string) error
 	StopJob(ctx context.Context, jobID string) error
 	ListTasks(ctx context.Context, offset, limit uint64) (task.TaskPage, error)
+	ListTasksByFilter(ctx context.Context, pm sdk.PageMetadata) (task.TaskPage, error)
 	UpdateTask(ctx context.Context, task task.Task) (task.Task, error)
 	DeleteTask(ctx context.Context, taskID string) error
 	StartTask(ctx context.Context, taskID string) error

manager/middleware/logging.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/absmach/propeller/manager"
 	"github.com/absmach/propeller/pkg/proplet"
+	"github.com/absmach/propeller/pkg/sdk"
 	"github.com/absmach/propeller/pkg/task"
 )
 
@@ -165,6 +166,25 @@ func (lm *loggingMiddleware) ListTasks(ctx context.Context, offset, limit uint64
 	return lm.svc.ListTasks(ctx, offset, limit)
 }
 
+func (lm *loggingMiddleware) ListTasksByFilter(ctx context.Context, pm sdk.PageMetadata) (resp task.TaskPage, err error) {
+	defer func(begin time.Time) {
+		args := []any{
+			slog.String("duration", time.Since(begin).String()),
+			slog.Uint64("offset", pm.Offset),
+			slog.Uint64("limit", pm.Limit),
+		}
+		if err != nil {
+			args = append(args, slog.Any("error", err))
+			lm.logger.Warn("List tasks by filter failed", args...)
+
+			return
+		}
+		lm.logger.Info("List tasks by filter completed successfully", args...)
+	}(time.Now())
+
+	return lm.svc.ListTasksByFilter(ctx, pm)
+}
+
 func (lm *loggingMiddleware) UpdateTask(ctx context.Context, t task.Task) (resp task.Task, err error) {
 	defer func(begin time.Time) {
 		args := []any{

manager/middleware/metrics.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/absmach/propeller/manager"
 	"github.com/absmach/propeller/pkg/proplet"
+	"github.com/absmach/propeller/pkg/sdk"
 	"github.com/absmach/propeller/pkg/task"
 	"github.com/go-kit/kit/metrics"
 )
@@ -89,6 +90,15 @@ func (mm *metricsMiddleware) ListTasks(ctx context.Context, offset, limit uint64
 	return mm.svc.ListTasks(ctx, offset, limit)
 }
 
+func (mm *metricsMiddleware) ListTasksByFilter(ctx context.Context, pm sdk.PageMetadata) (task.TaskPage, error) {
+	defer func(begin time.Time) {
+		mm.counter.With("method", "list-tasks-by-filter").Add(1)
+		mm.latency.With("method", "list-tasks-by-filter").Observe(time.Since(begin).Seconds())
+	}(time.Now())
+
+	return mm.svc.ListTasksByFilter(ctx, pm)
+}
+
 func (mm *metricsMiddleware) UpdateTask(ctx context.Context, t task.Task) (task.Task, error) {
 	defer func(begin time.Time) {
 		mm.counter.With("method", "update-task").Add(1)

manager/middleware/tracing.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/absmach/propeller/manager"
 	"github.com/absmach/propeller/pkg/proplet"
+	"github.com/absmach/propeller/pkg/sdk"
 	"github.com/absmach/propeller/pkg/task"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
@@ -90,6 +91,16 @@ func (tm *tracing) ListTasks(ctx context.Context, offset, limit uint64) (resp ta
 	return tm.svc.ListTasks(ctx, offset, limit)
 }
 
+func (tm *tracing) ListTasksByFilter(ctx context.Context, pm sdk.PageMetadata) (resp task.TaskPage, err error) {
+	ctx, span := tm.tracer.Start(ctx, "list-tasks-by-filter", trace.WithAttributes(
+		attribute.Int64("offset", int64(pm.Offset)),
+		attribute.Int64("limit", int64(pm.Limit)),
+	))
+	defer span.End()
+
+	return tm.svc.ListTasksByFilter(ctx, pm)
+}
+
 func (tm *tracing) UpdateTask(ctx context.Context, t task.Task) (resp task.Task, err error) {
 	ctx, span := tm.tracer.Start(ctx, "update-task", trace.WithAttributes(
 		attribute.String("id", resp.ID),