Merge pull request #900 from openmeterio/refactor/use-http-framework

refactor: use http framework for features

.vscode/settings.json

@@ -3,6 +3,8 @@
     "editor.formatOnSave": false
   },
   "go.testEnvVars": {
-    "TZ": "UTC"
+    "TZ": "UTC",
+    "POSTGRES_HOST": "127.0.0.1",
+    "OPENMETER_ADDRESS": "http://127.0.0.1:8888",
   }
 }

api/openapi.yaml

@@ -1260,7 +1260,7 @@ components:
       type: object
       # TODO: until https://github.com/deepmap/oapi-codegen/pull/1610 is not merged, we
       # need to copy the Problem schema here
-      # 
+      #
       # Afterwards the schema should be this
       #allOf:
       #- $ref: "#/components/schemas/Problem"
@@ -1446,7 +1446,7 @@ components:
     LedgerGrantBalance:
       # TODO: until https://github.com/deepmap/oapi-codegen/pull/1610 is not merged, the
       # generated go type will be invalid (it will be credit.Grant instead of GrantBalance).
-      # 
+      #
       # Given that we are not using this type as a return type of a call, this is just
       # an inconsistency we can ignore.
       x-go-type-import:
@@ -1464,7 +1464,7 @@ components:
     FeatureBalance:
       # TODO: until https://github.com/deepmap/oapi-codegen/pull/1610 is not merged, the
       # generated go type will be invalid (it will be credit.Feature instead of FeatureBalance).
-      # 
+      #
       # Given that we are not using this type as a return type of a call, this is just
       # an inconsistency we can ignore.
       x-go-type-import:

cmd/server/main.go

@@ -44,6 +44,7 @@ import (
 	"github.com/openmeterio/openmeter/internal/ingest/kafkaingest/serializer"
 	"github.com/openmeterio/openmeter/internal/meter"
 	"github.com/openmeterio/openmeter/internal/namespace"
+	"github.com/openmeterio/openmeter/internal/namespace/namespacedriver"
 	"github.com/openmeterio/openmeter/internal/server"
 	"github.com/openmeterio/openmeter/internal/server/authenticator"
 	"github.com/openmeterio/openmeter/internal/server/router"
@@ -269,7 +270,7 @@ func main() {
 	}
 	ingestHandler := ingestdriver.NewIngestEventsHandler(
 		ingestService.IngestEvents,
-		ingestdriver.StaticNamespaceDecoder(namespaceManager.GetDefaultNamespace()),
+		namespacedriver.StaticNamespaceDecoder(namespaceManager.GetDefaultNamespace()),
 		nil,
 		errorsx.NewContextHandler(errorsx.NewAppHandler(errorsx.NewSlogHandler(logger))),
 	)

config.example.yaml

@@ -13,8 +13,8 @@ telemetry:
 #     expiration: 768h # 32d
 
 # Entitlements
-#entitlements:
-#  enabled: true
+entitlements:
+  enabled: true
 
 # Consumer portal
 # portal:

internal/credit/connector.go

@@ -8,6 +8,7 @@ import (
 )
 
 type ListGrantsParams struct {
+	Namespace         string
 	LedgerIDs         []ulid.ULID
 	From              *time.Time
 	To                *time.Time
@@ -17,35 +18,49 @@ type ListGrantsParams struct {
 }
 
 type ListFeaturesParams struct {
+	Namespace       string
 	IncludeArchived bool
 }
 
 type ListLedgersParams struct {
-	Subjects []string
-	Offset   int
-	Limit    int
+	Namespace string
+	Subjects  []string
+	Offset    int
+	Limit     int
+}
+
+type NamespacedID struct {
+	Namespace string
+	ID        ulid.ULID
+}
+
+func NewNamespacedID(ns string, id ulid.ULID) NamespacedID {
+	return NamespacedID{
+		Namespace: ns,
+		ID:        id,
+	}
 }
 
 type Connector interface {
 	// Ledger
-	CreateLedger(ctx context.Context, namespace string, ledger Ledger) (Ledger, error)
-	ListLedgers(ctx context.Context, namespace string, params ListLedgersParams) ([]Ledger, error)
+	CreateLedger(ctx context.Context, ledger Ledger) (Ledger, error)
+	ListLedgers(ctx context.Context, params ListLedgersParams) ([]Ledger, error)
 
 	// Grant
-	CreateGrant(ctx context.Context, namespace string, grant Grant) (Grant, error)
-	VoidGrant(ctx context.Context, namespace string, grant Grant) (Grant, error)
-	ListGrants(ctx context.Context, namespace string, params ListGrantsParams) ([]Grant, error)
-	GetGrant(ctx context.Context, namespace string, id ulid.ULID) (Grant, error)
+	CreateGrant(ctx context.Context, grant Grant) (Grant, error)
+	VoidGrant(ctx context.Context, grant Grant) (Grant, error)
+	ListGrants(ctx context.Context, params ListGrantsParams) ([]Grant, error)
+	GetGrant(ctx context.Context, grantID NamespacedID) (Grant, error)
 
 	// Credit
-	GetBalance(ctx context.Context, namespace string, ledgerID ulid.ULID, cutline time.Time) (Balance, error)
-	GetHistory(ctx context.Context, namespace string, ledgerID ulid.ULID, from time.Time, to time.Time, limit int) (LedgerEntryList, error)
-	GetHighWatermark(ctx context.Context, namespace string, ledgerID ulid.ULID) (HighWatermark, error)
-	Reset(ctx context.Context, namespace string, reset Reset) (Reset, []Grant, error)
+	GetBalance(ctx context.Context, ledgerID NamespacedID, cutline time.Time) (Balance, error)
+	GetHistory(ctx context.Context, ledgerID NamespacedID, from time.Time, to time.Time, limit int) (LedgerEntryList, error)
+	GetHighWatermark(ctx context.Context, ledgerID NamespacedID) (HighWatermark, error)
+	Reset(ctx context.Context, reset Reset) (Reset, []Grant, error)
 
 	// Feature
-	CreateFeature(ctx context.Context, namespace string, feature Feature) (Feature, error)
-	DeleteFeature(ctx context.Context, namespace string, ledgerID ulid.ULID) error
-	ListFeatures(ctx context.Context, namespace string, params ListFeaturesParams) ([]Feature, error)
-	GetFeature(ctx context.Context, namespace string, ledgerID ulid.ULID) (Feature, error)
+	CreateFeature(ctx context.Context, feature Feature) (Feature, error)
+	DeleteFeature(ctx context.Context, featureID NamespacedID) error
+	ListFeatures(ctx context.Context, params ListFeaturesParams) ([]Feature, error)
+	GetFeature(ctx context.Context, featureID NamespacedID) (Feature, error)
 }

internal/credit/credit.go

@@ -96,6 +96,7 @@ func (GrantRolloverType) Values() (kinds []string) {
 
 // Reset is used to reset the balance of a specific subject.
 type Reset struct {
+	Namespace string `json:"-"`
 	// ID is the readonly identifies of a reset.
 	ID *ulid.ULID `json:"id,omitempty"`
 
@@ -113,6 +114,7 @@ func (c Reset) Render(w http.ResponseWriter, r *http.Request) error {
 
 // Grant is used to increase balance of specific subjects.
 type Grant struct {
+	Namespace string `json:"-"`
 	// ID is the readonly identifies of a grant.
 	ID *ulid.ULID `json:"id,omitempty"`
 
@@ -212,7 +214,7 @@ type HighWatermarBeforeError struct {
 }
 
 func (e *HighWatermarBeforeError) Error() string {
-	return fmt.Sprintf("ledger action for ledger %s.%s must be after highwatermark: %s", e.Namespace, e.LedgerID.String(), e.HighWatermark.Format(time.RFC3339))
+	return fmt.Sprintf("ledger action for ledger %s must be after highwatermark: %s", e.LedgerID.String(), e.HighWatermark.Format(time.RFC3339))
 }
 
 // LockErrNotObtainedError is returned when a lock cannot be obtained.
@@ -226,19 +228,17 @@ func (e *LockErrNotObtainedError) Error() string {
 }
 
 type LedgerAlreadyExistsError struct {
-	Namespace string
-	Ledger    Ledger
+	Ledger Ledger
 }
 
 func (e *LedgerAlreadyExistsError) Error() string {
-	return fmt.Sprintf("ledger (%s.%s) already exitst for subject %s", e.Namespace, e.Ledger.ID, e.Ledger.Subject)
+	return fmt.Sprintf("ledger %s already exitst for subject %s", e.Ledger.ID, e.Ledger.Subject)
 }
 
 type LedgerNotFoundError struct {
-	Namespace string
-	LedgerID  ulid.ULID
+	LedgerID ulid.ULID
 }
 
 func (e *LedgerNotFoundError) Error() string {
-	return fmt.Sprintf("ledger (%s.%s) not found", e.Namespace, e.LedgerID.String())
+	return fmt.Sprintf("ledger %s not found", e.LedgerID.String())
 }

internal/credit/creditdriver/balance.go

@@ -0,0 +1,63 @@
+package creditdriver
+
+import (
+	"context"
+	"net/http"
+	"time"
+
+	"github.com/openmeterio/openmeter/api"
+	"github.com/openmeterio/openmeter/internal/credit"
+	"github.com/openmeterio/openmeter/pkg/defaultx"
+	"github.com/openmeterio/openmeter/pkg/framework/commonhttp"
+	"github.com/openmeterio/openmeter/pkg/framework/transport/httptransport"
+)
+
+type GetLedgerBalaceHandlerParams struct {
+	LedgerID    api.LedgerID
+	QueryParams api.GetLedgerBalanceParams
+}
+
+type GetLedgerBalanceRequest struct {
+	LedgerID credit.NamespacedID
+	Cutline  time.Time
+}
+
+type GetLedgerBalanceHandler httptransport.HandlerWithArgs[GetLedgerBalanceRequest, credit.Balance, GetLedgerBalaceHandlerParams]
+
+func (b *builder) GetLedgerBalance() GetLedgerBalanceHandler {
+	return httptransport.NewHandlerWithArgs[GetLedgerBalanceRequest, credit.Balance, GetLedgerBalaceHandlerParams](
+		func(ctx context.Context, r *http.Request, queryIn GetLedgerBalaceHandlerParams) (GetLedgerBalanceRequest, error) {
+			ns, err := b.resolveNamespace(ctx)
+			if err != nil {
+				return GetLedgerBalanceRequest{}, err
+			}
+
+			return GetLedgerBalanceRequest{
+				LedgerID: credit.NewNamespacedID(ns, queryIn.LedgerID),
+				Cutline:  defaultx.WithDefault(queryIn.QueryParams.Time, time.Now()),
+			}, nil
+		},
+		func(ctx context.Context, request GetLedgerBalanceRequest) (credit.Balance, error) {
+			return b.CreditConnector.GetBalance(ctx, request.LedgerID, request.Cutline)
+		},
+		commonhttp.JSONResponseEncoder[credit.Balance],
+		httptransport.AppendOptions(
+			b.Options,
+			httptransport.WithOperationName("getLedgerBalance"),
+			httptransport.WithErrorEncoder(func(ctx context.Context, err error, w http.ResponseWriter) bool {
+				if _, ok := err.(*credit.HighWatermarBeforeError); ok {
+					commonhttp.NewHTTPError(http.StatusBadRequest, err).EncodeError(ctx, w)
+					return true
+				}
+				if _, ok := err.(*credit.LedgerNotFoundError); ok {
+					commonhttp.NewHTTPError(http.StatusNotFound, err).EncodeError(ctx, w)
+					return true
+				}
+				if _, ok := err.(*credit.LockErrNotObtainedError); ok {
+					commonhttp.NewHTTPError(http.StatusConflict, err).EncodeError(ctx, w)
+					return true
+				}
+				return false
+			}))...,
+	)
+}

internal/server/router/credit_limits.go → internal/credit/creditdriver/const.go

@@ -1,4 +1,4 @@
-package router
+package creditdriver
 
 const (
 	// DefaultLedgerQueryLimit specifies how many entries to return by default for credit related queries