billing/worker/subscriptionsync/service/reconciler/patch.go

package reconciler

import (
	"fmt"
	"slices"

	"github.com/alpacahq/alpacadecimal"
	"github.com/samber/lo"

	"github.com/openmeterio/openmeter/openmeter/billing"
	"github.com/openmeterio/openmeter/openmeter/billing/charges"
	"github.com/openmeterio/openmeter/openmeter/billing/worker/subscriptionsync/service/persistedstate"
	"github.com/openmeterio/openmeter/openmeter/billing/worker/subscriptionsync/service/reconciler/invoiceupdater"
	"github.com/openmeterio/openmeter/openmeter/billing/worker/subscriptionsync/service/targetstate"
	"github.com/openmeterio/openmeter/openmeter/productcatalog"
	"github.com/openmeterio/openmeter/pkg/featuregate"
	"github.com/openmeterio/openmeter/pkg/timeutil"
)

type PatchOperation string

const (
	PatchOperationCreate  PatchOperation = "create"
	PatchOperationDelete  PatchOperation = "delete"
	PatchOperationShrink  PatchOperation = "shrink"
	PatchOperationExtend  PatchOperation = "extend"
	PatchOperationProrate PatchOperation = "prorate"
)

type Patch interface {
	Operation() PatchOperation
	UniqueReferenceID() string
}

type InvoicePatch interface {
	Patch
	GetInvoicePatches() ([]invoiceupdater.Patch, error)
}

type InvoicePatchCollection interface {
	Patches() []InvoicePatch
	IsEmpty() bool
}

type ChargePatchCollection interface {
	Patches() charges.ApplyPatchesInput
	IsEmpty() bool
}

type PatchCollection interface {
	GetLineEngineType() billing.LineEngineType
	AddCreate(target targetstate.StateItem) error
	AddDelete(uniqueID string, existing persistedstate.Item) error
	AddShrink(uniqueID string, existing persistedstate.Item, target targetstate.StateItem) error
	AddExtend(existing persistedstate.Item, target targetstate.StateItem) error
	AddProrate(existing persistedstate.Item, target targetstate.StateItem, originalPeriod, targetPeriod timeutil.ClosedPeriod, originalAmount, targetAmount alpacadecimal.Decimal) error
}

type patchCollectionRouter struct {
	lineCollection             *lineInvoicePatchCollection
	hierarchyCollection        *lineHierarchyPatchCollection
	flatFeeChargeCollection    *flatFeeChargeCollection
	usageBasedChargeCollection *usageBasedChargeCollection
	creditThenInvoiceEnabled   bool
	creditsEnabled             bool
	featureGate                *featuregate.FeatureGateChecker
}

type patchCollectionRouterConfig struct {
	capacity                 int
	invoices                 persistedstate.Invoices
	creditThenInvoiceEnabled bool
	creditsEnabled           bool
	featureGate              *featuregate.FeatureGateChecker
}

func (c patchCollectionRouterConfig) Validate() error {
	if c.capacity <= 0 {
		return fmt.Errorf("capacity is required")
	}

	if c.invoices == nil {
		return fmt.Errorf("invoices is required")
	}

	if err := c.featureGate.Validate(); err != nil {
		return err
	}

	return nil
}

func newPatchCollectionRouter(cfg patchCollectionRouterConfig) (*patchCollectionRouter, error) {
	if err := cfg.Validate(); err != nil {
		return nil, err
	}

	lineCollection, err := newLineInvoicePatchCollection(cfg.invoices, cfg.capacity)
	if err != nil {
		return nil, fmt.Errorf("creating line collection: %w", err)
	}

	return &patchCollectionRouter{
		lineCollection:             lineCollection,
		hierarchyCollection:        newLineHierarchyPatchCollection(cfg.capacity),
		flatFeeChargeCollection:    newFlatFeeChargeCollection(cfg.capacity),
		usageBasedChargeCollection: newUsageBasedChargeCollection(cfg.capacity),
		creditThenInvoiceEnabled:   cfg.creditThenInvoiceEnabled,
		creditsEnabled:             cfg.creditsEnabled,
		featureGate:                cfg.featureGate,
	}, nil
}

func (c patchCollectionRouter) GetCollectionFor(item persistedstate.Item) (PatchCollection, error) {
	switch item.Type() {
	case persistedstate.ItemTypeInvoiceLine:
		return c.lineCollection, nil
	case persistedstate.ItemTypeInvoiceSplitLineGroup:
		return c.hierarchyCollection, nil
	case persistedstate.ItemTypeChargeFlatFee:
		return c.flatFeeChargeCollection, nil
	case persistedstate.ItemTypeChargeUsageBased:
		return c.usageBasedChargeCollection, nil
	default:
		return nil, fmt.Errorf("unsupported persisted item type: %s [id=%s]", item.Type(), item.ID())
	}
}

func (c patchCollectionRouter) isCreditsEnabled(ns string) (bool, error) {
	if !c.creditsEnabled {
		return false, nil
	}
	return c.featureGate.Enabled(ns, c.featureGate.Flags.Credits())
}

func (c patchCollectionRouter) ResolveDefaultCollection(target targetstate.StateItem) (PatchCollection, error) {
	enabled, err := c.isCreditsEnabled(target.SubscriptionItem.NamespacedID.Namespace)
	if err != nil {
		return nil, err
	}

	if !enabled {
		return c.lineCollection, nil
	}

	// If credit then invoice is not enabled, we return the lineCollection.
	if target.Subscription.SettlementMode == productcatalog.CreditThenInvoiceSettlementMode && !c.creditThenInvoiceEnabled {
		return c.lineCollection, nil
	}

	price := target.Spec.RateCard.AsMeta().Price
	if price == nil {
		// This should never happen as we are filtering for !IsBillable() targets in the filterInScopeLines function.
		return nil, fmt.Errorf("price is nil for target[%s]", target.UniqueID)
	}

	switch price.Type() {
	case productcatalog.FlatPriceType:
		return c.flatFeeChargeCollection, nil
	default:
		return c.usageBasedChargeCollection, nil
	}
}

func (c patchCollectionRouter) CollectInvoicePatches() []InvoicePatch {
	allPatches := slices.Concat(c.lineCollection.Patches(), c.hierarchyCollection.Patches())

	filtered := lo.Filter(allPatches, func(patch InvoicePatch, _ int) bool {
		return patch != nil
	})

	return filtered
}

func (c patchCollectionRouter) CollectChargePatches() (charges.ApplyPatchesInput, error) {
	return charges.ConcatenateApplyPatchesInputs(
		c.flatFeeChargeCollection.Patches(),
		c.usageBasedChargeCollection.Patches(),
	)
}