refactor: pricer billable period cleanup

openmeter/billing/service/invoice.go

@@ -8,7 +8,6 @@ import (
 
 	"github.com/oklog/ulid/v2"
 	"github.com/samber/lo"
-	"github.com/samber/mo"
 
 	"github.com/openmeterio/openmeter/openmeter/billing"
 	"github.com/openmeterio/openmeter/openmeter/billing/service/invoicecalc"
@@ -174,24 +173,14 @@ func (s *Service) recalculateGatheringInvoice(ctx context.Context, in recalculat
 		return invoice, fmt.Errorf("snapshotting lines: %w", err)
 	}
 
-	inScopeLineSvcs, err := lineservice.FromEntities(inScopeLines, featureMeters)
+	hasInvoicableLines, err := s.hasInvoicableLines(ctx, hasInvoicableLinesInput{
+		Invoice:            invoice,
+		AsOf:               now,
+		ProgressiveBilling: customerProfile.MergedProfile.WorkflowConfig.Invoicing.ProgressiveBilling,
+		FeatureMeters:      featureMeters,
+	})
 	if err != nil {
-		return invoice, fmt.Errorf("creating line services: %w", err)
-	}
-
-	hasInvoicableLines := mo.Option[bool]{}
-	for _, lineSvc := range inScopeLineSvcs {
-		period, err := lineSvc.CanBeInvoicedAsOf(lineservice.CanBeInvoicedAsOfInput{
-			AsOf:               now,
-			ProgressiveBilling: customerProfile.MergedProfile.WorkflowConfig.Invoicing.ProgressiveBilling,
-		})
-		if err != nil {
-			return invoice, fmt.Errorf("checking if can be invoiced: %w", err)
-		}
-
-		if period != nil {
-			hasInvoicableLines = mo.Some(true)
-		}
+		return invoice, fmt.Errorf("checking if has invoicable lines: %w", err)
 	}
 
 	invoice.QuantitySnapshotedAt = lo.ToPtr(now)
@@ -225,7 +214,8 @@ func (s *Service) recalculateGatheringInvoice(ctx context.Context, in recalculat
 	invoice.StatusDetails = billing.StandardInvoiceStatusDetails{
 		Immutable: false,
 		AvailableActions: billing.StandardInvoiceAvailableActions{
-			Invoice: lo.If(hasInvoicableLines.IsPresent(), &billing.StandardInvoiceAvailableActionInvoiceDetails{}).Else(nil),
+			Invoice: lo.If(hasInvoicableLines,
+				&billing.StandardInvoiceAvailableActionInvoiceDetails{}).Else(nil),
 		},
 	}
 
@@ -682,23 +672,19 @@ func (s Service) checkIfLinesAreInvoicable(ctx context.Context, invoice *billing
 			if err := line.Validate(); err != nil {
 				return fmt.Errorf("validating line[%s]: %w", line.ID, err)
 			}
-
-			lineSvc, err := lineservice.FromEntity(line, featureMeters)
-			if err != nil {
-				return fmt.Errorf("creating line service: %w", err)
-			}
-
-			period, err := lineSvc.CanBeInvoicedAsOf(lineservice.CanBeInvoicedAsOfInput{
-				AsOf:               lineSvc.InvoiceAt(),
+			period, err := lineservice.ResolveBillablePeriod(lineservice.ResolveBillablePeriodInput[*billing.StandardLine]{
+				Line:               line,
+				FeatureMeters:      featureMeters,
 				ProgressiveBilling: progressiveBilling,
+				AsOf:               line.InvoiceAt,
 			})
 			if err != nil {
-				return fmt.Errorf("checking if line[%s] can be invoiced: %w", lineSvc.ID(), err)
+				return fmt.Errorf("checking if line[%s] can be invoiced: %w", line.ID, err)
 			}
 
 			if period == nil {
 				return billing.ValidationError{
-					Err: fmt.Errorf("line[%s]: %w as of %s", lineSvc.ID(), billing.ErrInvoiceLinesNotBillable, lineSvc.Period().End),
+					Err: fmt.Errorf("line[%s]: %w as of %s", line.ID, billing.ErrInvoiceLinesNotBillable, line.Period.End),
 				}
 			}
 

openmeter/billing/service/lineservice/billableperiod.go

@@ -0,0 +1,151 @@
+package lineservice
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/openmeterio/openmeter/openmeter/billing"
+	"github.com/openmeterio/openmeter/openmeter/meter"
+	"github.com/openmeterio/openmeter/openmeter/productcatalog"
+	"github.com/openmeterio/openmeter/openmeter/streaming"
+	"github.com/openmeterio/openmeter/pkg/timeutil"
+)
+
+type PriceAccessor interface {
+	GetPrice() *productcatalog.Price
+	GetServicePeriod() timeutil.ClosedPeriod
+	GetFeatureKey() string
+}
+
+func IsPeriodEmptyConsideringTruncations(line PriceAccessor) (bool, error) {
+	price := line.GetPrice()
+	if price == nil {
+		return false, fmt.Errorf("price is nil")
+	}
+
+	if price.Type() == productcatalog.FlatPriceType {
+		// Flat prices are always billable even if the period is empty
+		return false, nil
+	}
+
+	return line.GetServicePeriod().Truncate(streaming.MinimumWindowSizeDuration).IsEmpty(), nil
+}
+
+type GetLinesWithBillablePeriodsInput[T PriceAccessor] struct {
+	AsOf               time.Time
+	ProgressiveBilling bool
+	Lines              []T
+	FeatureMeters      billing.FeatureMeters
+}
+
+type LineWithBillablePeriod[T PriceAccessor] struct {
+	Line           T
+	BillablePeriod timeutil.ClosedPeriod
+}
+
+func GetLinesWithBillablePeriods[T PricerCanBeInvoicedAsOfAccessor](in GetLinesWithBillablePeriodsInput[T]) ([]LineWithBillablePeriod[T], error) {
+	out := make([]LineWithBillablePeriod[T], 0, len(in.Lines))
+	for _, line := range in.Lines {
+		billablePeriod, err := ResolveBillablePeriod(ResolveBillablePeriodInput[T]{
+			AsOf:               in.AsOf,
+			ProgressiveBilling: in.ProgressiveBilling,
+			Line:               line,
+			FeatureMeters:      in.FeatureMeters,
+		})
+		if err != nil {
+			return nil, fmt.Errorf("line[%s]: %w", line.GetID(), err)
+		}
+
+		if billablePeriod == nil {
+			continue
+		}
+
+		out = append(out, LineWithBillablePeriod[T]{
+			Line:           line,
+			BillablePeriod: *billablePeriod,
+		})
+	}
+
+	return out, nil
+}
+
+type ResolveBillablePeriodInput[T PricerCanBeInvoicedAsOfAccessor] struct {
+	AsOf               time.Time
+	ProgressiveBilling bool
+	Line               T
+	FeatureMeters      billing.FeatureMeters
+}
+
+func ResolveBillablePeriod[T PricerCanBeInvoicedAsOfAccessor](in ResolveBillablePeriodInput[T]) (*timeutil.ClosedPeriod, error) {
+	pricer, err := newPricerFor(in.Line)
+	if err != nil {
+		return nil, err
+	}
+
+	price := in.Line.GetPrice()
+	if price == nil {
+		return nil, fmt.Errorf("price is nil")
+	}
+
+	meterTypeAllowsProgressiveBilling := false
+	if price.Type() != productcatalog.FlatPriceType && in.ProgressiveBilling {
+		isDependingOnIncreaseOnlyMeters, err := isDependingOnIncreaseOnlyMeters(CanBeInvoicedAsOfInput{
+			AsOf:               in.AsOf,
+			ProgressiveBilling: in.ProgressiveBilling,
+			Line:               in.Line,
+			FeatureMeters:      in.FeatureMeters,
+		})
+		if err != nil {
+			return nil, err
+		}
+
+		meterTypeAllowsProgressiveBilling = isDependingOnIncreaseOnlyMeters
+	}
+
+	// Force disable progressive billing if the meter type does not allow it
+	if !meterTypeAllowsProgressiveBilling {
+		in.ProgressiveBilling = false
+	}
+
+	billablePeriod, err := pricer.CanBeInvoicedAsOf(CanBeInvoicedAsOfInput{
+		AsOf:               in.AsOf,
+		ProgressiveBilling: in.ProgressiveBilling,
+		Line:               in.Line,
+		FeatureMeters:      in.FeatureMeters,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return billablePeriod, nil
+}
+
+// isDependingOnIncreaseOnlyMeters checks if the line is depending on meters that can decrease the totals over time
+// (note: this is somewhat of a lie, as we can input negative values in events, which will have the same effect)
+func isDependingOnIncreaseOnlyMeters(in CanBeInvoicedAsOfInput) (bool, error) {
+	featureKey := in.Line.GetFeatureKey()
+	if featureKey == "" {
+		return false, fmt.Errorf("feature key is required")
+	}
+
+	// Let's check if the underlying meter can be billed in a progressive manner
+	featureMeter, err := in.FeatureMeters.Get(featureKey, true)
+	if err != nil {
+		return false, err
+	}
+
+	if featureMeter.Meter == nil {
+		return false, fmt.Errorf("meter is nil for feature[%s]", featureKey)
+	}
+
+	meterEntity := *featureMeter.Meter
+
+	switch meterEntity.Aggregation {
+	case meter.MeterAggregationSum, meter.MeterAggregationCount,
+		meter.MeterAggregationMax, meter.MeterAggregationUniqueCount:
+		return true, nil
+	default:
+		// Other types need to be billed in arrears truncated by window size
+		return false, nil
+	}
+}

openmeter/billing/service/lineservice/pricedynamic.go

@@ -9,7 +9,7 @@ import (
 )
 
 type dynamicPricer struct {
-	ProgressiveBillingPricer
+	ProgressiveBillingMeteredPricer
 }
 
 var _ Pricer = (*dynamicPricer)(nil)

openmeter/billing/service/lineservice/priceflat.go

@@ -5,15 +5,15 @@ import (
 	"slices"
 
 	"github.com/alpacahq/alpacadecimal"
+	"github.com/samber/lo"
 
 	"github.com/openmeterio/openmeter/openmeter/billing"
 	"github.com/openmeterio/openmeter/openmeter/productcatalog"
+	"github.com/openmeterio/openmeter/openmeter/streaming"
+	"github.com/openmeterio/openmeter/pkg/timeutil"
 )
 
-type flatPricer struct {
-	// TODO[later]: when we introduce full pro-rating support this should be ProgressiveBillingPricer
-	NonProgressiveBillingPricer
-}
+type flatPricer struct{}
 
 var _ Pricer = (*flatPricer)(nil)
 
@@ -60,3 +60,22 @@ func (p flatPricer) Calculate(l PricerCalculateInput) (newDetailedLinesInput, er
 
 	return nil, nil
 }
+
+func (p flatPricer) CanBeInvoicedAsOf(in CanBeInvoicedAsOfInput) (*timeutil.ClosedPeriod, error) {
+	if in.Line.GetSplitLineGroupID() != nil {
+		return nil, billing.ValidationError{
+			Err: billing.ErrInvoiceProgressiveBillingNotSupported,
+		}
+	}
+
+	// For the flat prices they are always billable but the invoiceAt signifies when the line should be
+	// actually billed.
+	invoiceAtTruncated := in.Line.GetInvoiceAt().Truncate(streaming.MinimumWindowSizeDuration)
+	asOfTruncated := in.AsOf.Truncate(streaming.MinimumWindowSizeDuration)
+
+	if invoiceAtTruncated.After(asOfTruncated) {
+		return nil, nil
+	}
+
+	return lo.ToPtr(in.Line.GetServicePeriod()), nil
+}

openmeter/billing/service/lineservice/pricegraduatedtiered.go

@@ -10,7 +10,7 @@ import (
 )
 
 type graduatedTieredPricer struct {
-	ProgressiveBillingPricer
+	ProgressiveBillingMeteredPricer
 }
 
 var _ Pricer = (*graduatedTieredPricer)(nil)

openmeter/billing/service/lineservice/pricemutate.go

@@ -1,6 +1,6 @@
 package lineservice
 
-import "time"
+import "github.com/openmeterio/openmeter/pkg/timeutil"
 
 type priceMutator struct {
 	PreCalculation  []PreCalculationMutator
@@ -45,6 +45,6 @@ func (p *priceMutator) Calculate(l PricerCalculateInput) (newDetailedLinesInput,
 	return newDetailedLines, nil
 }
 
-func (p *priceMutator) CanBeInvoicedAsOf(l usageBasedLine, asOf time.Time) (bool, error) {
-	return p.Pricer.CanBeInvoicedAsOf(l, asOf)
+func (p *priceMutator) CanBeInvoicedAsOf(in CanBeInvoicedAsOfInput) (*timeutil.ClosedPeriod, error) {
+	return p.Pricer.CanBeInvoicedAsOf(in)
 }

openmeter/billing/service/lineservice/pricepackage.go

@@ -9,7 +9,7 @@ import (
 )
 
 type packagePricer struct {
-	ProgressiveBillingPricer
+	ProgressiveBillingMeteredPricer
 }
 
 var _ Pricer = (*packagePricer)(nil)