meter/httphandler/query.go

package httpdriver

import (
	"context"
	"fmt"
	"net/http"
	"time"

	"github.com/samber/lo"

	"github.com/openmeterio/openmeter/api"
	"github.com/openmeterio/openmeter/openmeter/meter"
	"github.com/openmeterio/openmeter/openmeter/streaming"
	"github.com/openmeterio/openmeter/pkg/framework/commonhttp"
	"github.com/openmeterio/openmeter/pkg/framework/transport/httptransport"
)

type (
	ListSubjectsResponse = *[]string
	ListSubjectsHandler  httptransport.HandlerWithArgs[ListSubjectsRequest, ListSubjectsResponse, ListSubjectsParams]
)

type ListSubjectsParams struct {
	IdOrSlug string
	From     *time.Time
	To       *time.Time
}

type ListSubjectsRequest struct {
	namespace string
	idOrSlug  string
	from      *time.Time
	to        *time.Time
}

// ListSubjects returns a handler for query meter.
func (h *handler) ListSubjects() ListSubjectsHandler {
	return httptransport.NewHandlerWithArgs(
		func(ctx context.Context, r *http.Request, params ListSubjectsParams) (ListSubjectsRequest, error) {
			ns, err := h.resolveNamespace(ctx)
			if err != nil {
				return ListSubjectsRequest{}, err
			}

			return ListSubjectsRequest{
				namespace: ns,
				idOrSlug:  params.IdOrSlug,
				from:      params.From,
				to:        params.To,
			}, nil
		},
		func(ctx context.Context, request ListSubjectsRequest) (ListSubjectsResponse, error) {
			meter, err := h.meterService.GetMeterByIDOrSlug(ctx, meter.GetMeterInput{
				Namespace: request.namespace,
				IDOrSlug:  request.idOrSlug,
			})
			if err != nil {
				return nil, fmt.Errorf("failed to get meter: %w", err)
			}

			subjectKeys, err := h.streaming.ListSubjects(ctx, streaming.ListSubjectsParams{
				Namespace: request.namespace,
				Meter:     &meter,
				From:      request.from,
				To:        request.to,
			})
			if err != nil {
				return nil, fmt.Errorf("failed to list subjects: %w", err)
			}

			return &subjectKeys, nil
		},
		commonhttp.JSONResponseEncoderWithStatus[ListSubjectsResponse](http.StatusOK),
		httptransport.AppendOptions(
			h.options,
			httptransport.WithOperationName("queryMeter"),
		)...,
	)
}

type (
	QueryMeterResponse = *api.MeterQueryResult
	QueryMeterHandler  httptransport.HandlerWithArgs[QueryMeterRequest, QueryMeterResponse, QueryMeterParams]
)

type QueryMeterParams struct {
	api.QueryMeterParams
	IdOrSlug string
}

type QueryMeterRequest struct {
	namespace string
	idOrSlug  string
	params    api.QueryMeterParams
}

// QueryMeter returns a handler for query meter.
func (h *handler) QueryMeter() QueryMeterHandler {
	return httptransport.NewHandlerWithArgs(
		func(ctx context.Context, r *http.Request, params QueryMeterParams) (QueryMeterRequest, error) {
			ns, err := h.resolveNamespace(ctx)
			if err != nil {
				return QueryMeterRequest{}, err
			}

			return QueryMeterRequest{
				namespace: ns,
				idOrSlug:  params.IdOrSlug,
				params:    params.QueryMeterParams,
			}, nil
		},
		func(ctx context.Context, request QueryMeterRequest) (QueryMeterResponse, error) {
			meter, err := h.meterService.GetMeterByIDOrSlug(ctx, meter.GetMeterInput{
				Namespace: request.namespace,
				IDOrSlug:  request.idOrSlug,
			})
			if err != nil {
				return nil, fmt.Errorf("failed to get meter: %w", err)
			}

			params, err := h.toQueryParamsFromRequest(
				ctx,
				meter,
				// Convert the POST request body to a GET request params
				ToRequestFromQueryParamsPOSTBody(request.params),
			)
			if err != nil {
				return nil, fmt.Errorf("failed to construct query meter params: %w", err)
			}

			rows, err := h.streaming.QueryMeter(ctx, request.namespace, meter, params)
			if err != nil {
				return nil, fmt.Errorf("failed to query meter: %w", err)
			}

			response := ToAPIMeterQueryResult(request.params.From, request.params.To, request.params.WindowSize, rows)

			return &response, nil
		},
		commonhttp.JSONResponseEncoderWithStatus[QueryMeterResponse](http.StatusOK),
		httptransport.AppendOptions(
			h.options,
			httptransport.WithOperationName("queryMeter"),
		)...,
	)
}

type QueryMeterPostRequest struct {
	namespace string
	idOrSlug  string
	params    api.QueryMeterPostJSONRequestBody
}

type (
	QueryMeterPostParams   = string // meterIdOrSlug
	QueryMeterPostResponse = QueryMeterResponse
	QueryMeterPostHandler  httptransport.HandlerWithArgs[QueryMeterPostRequest, QueryMeterPostResponse, QueryMeterPostParams]
)

// QueryMeterPost returns a handler for query meter via POST.
func (h *handler) QueryMeterPost() QueryMeterPostHandler {
	return httptransport.NewHandlerWithArgs(
		func(ctx context.Context, r *http.Request, meterIdOrSlug QueryMeterPostParams) (QueryMeterPostRequest, error) {
			ns, err := h.resolveNamespace(ctx)
			if err != nil {
				return QueryMeterPostRequest{}, err
			}

			var request api.QueryMeterPostJSONRequestBody
			if err := commonhttp.JSONRequestBodyDecoder(r, &request); err != nil {
				return QueryMeterPostRequest{}, fmt.Errorf("failed to decode request body: %w", err)
			}

			return QueryMeterPostRequest{
				namespace: ns,
				idOrSlug:  meterIdOrSlug,
				params:    request,
			}, nil
		},
		func(ctx context.Context, request QueryMeterPostRequest) (QueryMeterPostResponse, error) {
			meter, err := h.meterService.GetMeterByIDOrSlug(ctx, meter.GetMeterInput{
				Namespace: request.namespace,
				IDOrSlug:  request.idOrSlug,
			})
			if err != nil {
				return nil, fmt.Errorf("failed to get meter: %w", err)
			}

			params, err := h.toQueryParamsFromRequest(ctx, meter, request.params)
			if err != nil {
				return nil, fmt.Errorf("failed to construct query meter params: %w", err)
			}

			rows, err := h.streaming.QueryMeter(ctx, request.namespace, meter, params)
			if err != nil {
				return nil, fmt.Errorf("failed to query meter: %w", err)
			}

			response := ToAPIMeterQueryResult(request.params.From, request.params.To, request.params.WindowSize, rows)

			return &response, nil
		},
		commonhttp.JSONResponseEncoderWithStatus[QueryMeterPostResponse](http.StatusOK),
		httptransport.AppendOptions(
			h.options,
			httptransport.WithOperationName("queryMeterPost"),
		)...,
	)
}

type (
	ListGroupByValuesResponse = *[]string
	ListGroupByValuesHandler  httptransport.HandlerWithArgs[ListGroupByValuesRequest, ListGroupByValuesResponse, ListGroupByValuesParams]
)

type ListGroupByValuesParams struct {
	IdOrSlug   string
	GroupByKey string
	From       *time.Time
	To         *time.Time
	Search     *string
}

type ListGroupByValuesRequest struct {
	namespace  string
	idOrSlug   string
	groupByKey string
	from       *time.Time
	to         *time.Time
	search     *string
}

// ListGroupByValues returns a handler for list group by values.
func (h *handler) ListGroupByValues() ListGroupByValuesHandler {
	return httptransport.NewHandlerWithArgs(
		func(ctx context.Context, r *http.Request, params ListGroupByValuesParams) (ListGroupByValuesRequest, error) {
			ns, err := h.resolveNamespace(ctx)
			if err != nil {
				return ListGroupByValuesRequest{}, err
			}

			// Set default to last 24 hours
			if params.From == nil && params.To == nil {
				params.From = lo.ToPtr(time.Now().Add(-time.Hour * 24))
			}

			return ListGroupByValuesRequest{
				namespace:  ns,
				idOrSlug:   params.IdOrSlug,
				groupByKey: params.GroupByKey,
				from:       params.From,
				to:         params.To,
				search:     params.Search,
			}, nil
		},
		func(ctx context.Context, request ListGroupByValuesRequest) (ListGroupByValuesResponse, error) {
			meter, err := h.meterService.GetMeterByIDOrSlug(ctx, meter.GetMeterInput{
				Namespace: request.namespace,
				IDOrSlug:  request.idOrSlug,
			})
			if err != nil {
				return nil, fmt.Errorf("failed to get meter: %w", err)
			}

			groupByValues, err := h.streaming.ListGroupByValues(ctx, streaming.ListGroupByValuesParams{
				Namespace:  request.namespace,
				Meter:      meter,
				GroupByKey: request.groupByKey,
				From:       request.from,
				To:         request.to,
				Search:     request.search,
			})
			if err != nil {
				return nil, fmt.Errorf("failed to list group by values: %w", err)
			}

			return &groupByValues, nil
		},
		commonhttp.JSONResponseEncoderWithStatus[ListGroupByValuesResponse](http.StatusOK),
		httptransport.AppendOptions(
			h.options,
			httptransport.WithOperationName("listGroupByValues"),
		)...,
	)
}