diff --git a/app/common/billing.go b/app/common/billing.go index 27d457a38f..9de1c300f1 100644 --- a/app/common/billing.go +++ b/app/common/billing.go @@ -16,6 +16,7 @@ import ( "github.com/openmeterio/openmeter/openmeter/billing/charges/flatfee" "github.com/openmeterio/openmeter/openmeter/billing/charges/usagebased" chargesworkeradvance "github.com/openmeterio/openmeter/openmeter/billing/charges/worker/advance" + billinglineengine "github.com/openmeterio/openmeter/openmeter/billing/lineengine" "github.com/openmeterio/openmeter/openmeter/billing/rating" billingratingservice "github.com/openmeterio/openmeter/openmeter/billing/rating/service" billingsequence "github.com/openmeterio/openmeter/openmeter/billing/sequence" @@ -48,9 +49,10 @@ import ( // BillingRegistry bundles the billing and charges services. External callers that need // billing or charges should depend on BillingRegistry rather than individual services. type BillingRegistry struct { - Billing billing.Service - Sequence billingsequence.Service - Charges *ChargesRegistry + Billing billing.Service + LegacyBillingLineEngine *billinglineengine.Engine + Sequence billingsequence.Service + Charges *ChargesRegistry } func (r BillingRegistry) ChargesServiceOrNil() charges.Service { @@ -109,6 +111,22 @@ func NewBillingSequenceService( }) } +func NewLegacyBillingLineEngine( + billingAdapter billing.Adapter, + billingRatingService rating.Service, + featureConnector feature.FeatureConnector, + streamingConnector streaming.Connector, + billingConfig config.BillingConfiguration, +) (*billinglineengine.Engine, error) { + return billinglineengine.New(billinglineengine.Config{ + SplitLineGroupAdapter: billingAdapter, + RatingService: billingRatingService, + FeatureService: featureConnector, + StreamingConnector: streamingConnector, + MaxParallelQuantitySnapshots: billingConfig.MaxParallelQuantitySnapshots, + }) +} + // newBillingService creates the billing service and registers validators/hooks. // Downstream consumers should use BillingRegistry. func newBillingService( @@ -116,11 +134,11 @@ func newBillingService( appService app.Service, billingAdapter billing.Adapter, billingRatingService rating.Service, + legacyBillingLineEngine *billinglineengine.Engine, sequenceService billingsequence.Service, customerService customer.Service, featureConnector feature.FeatureConnector, meterService meter.Service, - streamingConnector streaming.Connector, eventPublisher eventbus.Publisher, billingConfig config.BillingConfiguration, subscriptionServices SubscriptionServiceWithWorkflow, @@ -130,20 +148,19 @@ func newBillingService( taxCodeService taxcode.Service, ) (billing.Service, error) { service, err := billingservice.New(billingservice.Config{ - Adapter: billingAdapter, - SequenceService: sequenceService, - RatingService: billingRatingService, - AppService: appService, - CustomerService: customerService, - TaxCodeService: taxCodeService, - FeatureService: featureConnector, - Logger: logger, - MeterService: meterService, - StreamingConnector: streamingConnector, - Publisher: eventPublisher, - AdvancementStrategy: billingConfig.AdvancementStrategy, - FSNamespaceLockdown: fsConfig.NamespaceLockdown, - MaxParallelQuantitySnapshots: billingConfig.MaxParallelQuantitySnapshots, + Adapter: billingAdapter, + SequenceService: sequenceService, + RatingService: billingRatingService, + LegacyBillingLineEngine: legacyBillingLineEngine, + AppService: appService, + CustomerService: customerService, + TaxCodeService: taxCodeService, + FeatureService: featureConnector, + Logger: logger, + MeterService: meterService, + Publisher: eventPublisher, + AdvancementStrategy: billingConfig.AdvancementStrategy, + FSNamespaceLockdown: fsConfig.NamespaceLockdown, }) if err != nil { return nil, err @@ -184,16 +201,27 @@ func NewBillingRegistry( return BillingRegistry{}, err } + legacyBillingLineEngine, err := NewLegacyBillingLineEngine( + billingAdapter, + billingRatingService, + featureConnector, + streamingConnector, + billingConfig, + ) + if err != nil { + return BillingRegistry{}, err + } + billingService, err := newBillingService( logger, appService, billingAdapter, billingRatingService, + legacyBillingLineEngine, sequenceService, customerService, featureConnector, meterService, - streamingConnector, eventPublisher, billingConfig, subscriptionServices, @@ -233,9 +261,10 @@ func NewBillingRegistry( } billingRegistry := BillingRegistry{ - Billing: billingService, - Sequence: sequenceService, - Charges: chargesRegistry, + Billing: billingService, + LegacyBillingLineEngine: legacyBillingLineEngine, + Sequence: sequenceService, + Charges: chargesRegistry, } // Hook registration @@ -330,6 +359,7 @@ func NewBillingSubscriptionSyncService(logger *slog.Logger, subsServices Subscri return subscriptionsyncservice.New(subscriptionsyncservice.Config{ SubscriptionService: subsServices.Service, BillingService: billingRegistry.Billing, + LegacyBillingLineEngine: billingRegistry.LegacyBillingLineEngine, ChargesService: billingRegistry.ChargesServiceOrNil(), SubscriptionSyncAdapter: subscriptionSyncAdapter, FeatureFlags: subscriptionsyncservice.FeatureFlags{ diff --git a/openmeter/billing/lineengine/engine.go b/openmeter/billing/lineengine/engine.go index f1a77bd49f..c7d75319a7 100644 --- a/openmeter/billing/lineengine/engine.go +++ b/openmeter/billing/lineengine/engine.go @@ -9,6 +9,8 @@ import ( "github.com/openmeterio/openmeter/openmeter/billing" "github.com/openmeterio/openmeter/openmeter/billing/rating" "github.com/openmeterio/openmeter/openmeter/productcatalog" + "github.com/openmeterio/openmeter/openmeter/productcatalog/feature" + "github.com/openmeterio/openmeter/openmeter/streaming" "github.com/openmeterio/openmeter/pkg/clock" "github.com/openmeterio/openmeter/pkg/equal" "github.com/openmeterio/openmeter/pkg/slicesx" @@ -20,9 +22,11 @@ var ( ) type Config struct { - SplitLineGroupAdapter SplitLineGroupAdapter - QuantitySnapshotter QuantitySnapshotter - RatingService rating.Service + SplitLineGroupAdapter SplitLineGroupAdapter + RatingService rating.Service + FeatureService feature.FeatureConnector + StreamingConnector streaming.Connector + MaxParallelQuantitySnapshots int } func (c Config) Validate() error { @@ -30,21 +34,31 @@ func (c Config) Validate() error { return fmt.Errorf("split line group adapter is required") } - if c.QuantitySnapshotter == nil { - return fmt.Errorf("quantity snapshotter is required") - } - if c.RatingService == nil { return fmt.Errorf("rating service is required") } + if c.FeatureService == nil { + return fmt.Errorf("feature service is required") + } + + if c.StreamingConnector == nil { + return fmt.Errorf("streaming connector is required") + } + + if c.MaxParallelQuantitySnapshots < 1 { + return fmt.Errorf("max parallel quantity snapshots must be greater than 0") + } + return nil } type Engine struct { - adapter SplitLineGroupAdapter - quantitySnapshotter QuantitySnapshotter - ratingService rating.Service + adapter SplitLineGroupAdapter + ratingService rating.Service + featureService feature.FeatureConnector + streamingConnector streaming.Connector + maxParallelQuantitySnapshots int } func New(config Config) (*Engine, error) { @@ -53,9 +67,11 @@ func New(config Config) (*Engine, error) { } return &Engine{ - adapter: config.SplitLineGroupAdapter, - quantitySnapshotter: config.QuantitySnapshotter, - ratingService: config.RatingService, + adapter: config.SplitLineGroupAdapter, + ratingService: config.RatingService, + featureService: config.FeatureService, + streamingConnector: config.StreamingConnector, + maxParallelQuantitySnapshots: config.MaxParallelQuantitySnapshots, }, nil } @@ -79,7 +95,7 @@ func (e *Engine) OnCollectionCompleted(ctx context.Context, input billing.OnColl return input.Lines, nil } - if err := e.quantitySnapshotter.SnapshotLineQuantities(ctx, input.Invoice, input.Lines); err != nil { + if err := e.SnapshotLineQuantities(ctx, input.Invoice, input.Lines); err != nil { if _, isInvalidDatabaseState := lo.ErrorsAs[*billing.ErrSnapshotInvalidDatabaseState](err); isInvalidDatabaseState { return nil, billing.ValidationIssue{ Severity: billing.ValidationIssueSeverityCritical, @@ -174,7 +190,7 @@ func (e *Engine) snapshotManualStandardLineOverrideIfNeeded(ctx context.Context, return nil, fmt.Errorf("getting standard line: %w", err) } - if err := e.quantitySnapshotter.SnapshotLineQuantities(ctx, standardInvoice, billing.StandardLines{&standardLine}); err != nil { + if err := e.SnapshotLineQuantities(ctx, standardInvoice, billing.StandardLines{&standardLine}); err != nil { return nil, fmt.Errorf("snapshotting line quantity: %w", err) } diff --git a/openmeter/billing/service/quantitysnapshot.go b/openmeter/billing/lineengine/quantitysnapshot.go similarity index 65% rename from openmeter/billing/service/quantitysnapshot.go rename to openmeter/billing/lineengine/quantitysnapshot.go index d7feba5ea3..107f985de6 100644 --- a/openmeter/billing/service/quantitysnapshot.go +++ b/openmeter/billing/lineengine/quantitysnapshot.go @@ -1,4 +1,4 @@ -package billingservice +package lineengine import ( "context" @@ -15,21 +15,39 @@ import ( "github.com/openmeterio/openmeter/openmeter/meter" "github.com/openmeterio/openmeter/openmeter/productcatalog/feature" "github.com/openmeterio/openmeter/openmeter/streaming" + "github.com/openmeterio/openmeter/pkg/ref" ) -func (s *Service) SnapshotLineQuantity(ctx context.Context, input billing.SnapshotLineQuantityInput) (*billing.StandardLine, error) { +type SnapshotLineQuantityInput struct { + Invoice *billing.StandardInvoice + Line *billing.StandardLine +} + +func (i SnapshotLineQuantityInput) Validate() error { + var errs []error + if i.Invoice == nil { + errs = append(errs, errors.New("invoice is required")) + } + if i.Line == nil { + errs = append(errs, errors.New("line is required")) + } + + return errors.Join(errs...) +} + +func (e *Engine) SnapshotLineQuantity(ctx context.Context, input SnapshotLineQuantityInput) (*billing.StandardLine, error) { if err := input.Validate(); err != nil { return nil, billing.ValidationError{ Err: err, } } - featureMeters, err := s.resolveFeatureMeters(ctx, input.Invoice.Namespace, billing.StandardLines{input.Line}) + featureMeters, err := e.resolveFeatureMeters(ctx, input.Invoice.Namespace, billing.StandardLines{input.Line}) if err != nil { return nil, fmt.Errorf("line[%s]: %w", input.Line.ID, err) } - err = s.snapshotLineQuantity(ctx, input.Invoice.Customer, input.Line, featureMeters) + err = e.snapshotLineQuantity(ctx, input.Invoice.Customer, input.Line, featureMeters) if err != nil { return nil, err } @@ -37,26 +55,58 @@ func (s *Service) SnapshotLineQuantity(ctx context.Context, input billing.Snapsh return input.Line, nil } -func (s *Service) SnapshotLineQuantities(ctx context.Context, invoice billing.StandardInvoice, lines billing.StandardLines) error { - featureMeters, err := s.resolveFeatureMeters(ctx, invoice.Namespace, lines) +func (e *Engine) SnapshotLineQuantities(ctx context.Context, invoice billing.StandardInvoice, lines billing.StandardLines) error { + featureMeters, err := e.resolveFeatureMeters(ctx, invoice.Namespace, lines) if err != nil { return fmt.Errorf("resolving feature meters: %w", err) } - if err := s.snapshotLineQuantitiesInParallel(ctx, invoice.Customer, lines, featureMeters); err != nil { + if err := e.snapshotLineQuantitiesInParallel(ctx, invoice.Customer, lines, featureMeters); err != nil { return fmt.Errorf("snapshotting lines: %w", err) } return nil } -func (s *Service) snapshotMeteredLineQuantity(ctx context.Context, line *billing.StandardLine, customer billing.InvoiceCustomer, featureMeters feature.FeatureMeters) error { +func (e *Engine) resolveFeatureMeters(ctx context.Context, namespace string, lines billing.StandardLines) (feature.FeatureMeters, error) { + keys, err := lines.GetReferencedFeatureKeys() + if err != nil { + return nil, fmt.Errorf("getting referenced feature keys: %w", err) + } + + featureMeters, err := e.featureService.ResolveFeatureMeters(ctx, namespace, lo.Map(keys, func(key string, _ int) ref.IDOrKey { + return ref.IDOrKey{Key: key} + })...) + if err != nil { + return nil, fmt.Errorf("resolving feature meters: %w", err) + } + + return featureMetersErrorWrapper{featureMeters}, nil +} + +// featureMetersErrorWrapper returns ErrSnapshotInvalidDatabaseState when a feature meter is unavailable during snapshotting. +type featureMetersErrorWrapper struct { + feature.FeatureMeters +} + +func (w featureMetersErrorWrapper) Get(featureKey string, requireMeter bool) (feature.FeatureMeter, error) { + featureMeter, err := w.FeatureMeters.Get(featureKey, requireMeter) + if err != nil { + return feature.FeatureMeter{}, &billing.ErrSnapshotInvalidDatabaseState{ + Err: err, + } + } + + return featureMeter, nil +} + +func (e *Engine) snapshotMeteredLineQuantity(ctx context.Context, line *billing.StandardLine, customer billing.InvoiceCustomer, featureMeters feature.FeatureMeters) error { featureMeter, err := featureMeters.Get(line.UsageBased.FeatureKey, true) if err != nil { return err } - usage, err := s.getFeatureUsage(ctx, + usage, err := e.getFeatureUsage(ctx, getFeatureUsageInput{ Line: line, Feature: featureMeter.Feature, @@ -76,7 +126,7 @@ func (s *Service) snapshotMeteredLineQuantity(ctx context.Context, line *billing return nil } -func (s *Service) snapshotFlatPriceLineQuantity(_ context.Context, line *billing.StandardLine) error { +func (e *Engine) snapshotFlatPriceLineQuantity(_ context.Context, line *billing.StandardLine) error { line.UsageBased.MeteredQuantity = lo.ToPtr(alpacadecimal.NewFromInt(1)) line.UsageBased.Quantity = lo.ToPtr(alpacadecimal.NewFromInt(1)) line.UsageBased.PreLinePeriodQuantity = lo.ToPtr(alpacadecimal.Zero) @@ -84,16 +134,16 @@ func (s *Service) snapshotFlatPriceLineQuantity(_ context.Context, line *billing return nil } -func (s *Service) snapshotLineQuantity(ctx context.Context, customer billing.InvoiceCustomer, line *billing.StandardLine, featureMeters feature.FeatureMeters) error { +func (e *Engine) snapshotLineQuantity(ctx context.Context, customer billing.InvoiceCustomer, line *billing.StandardLine, featureMeters feature.FeatureMeters) error { if !line.DependsOnMeteredQuantity() { - return s.snapshotFlatPriceLineQuantity(ctx, line) + return e.snapshotFlatPriceLineQuantity(ctx, line) } - return s.snapshotMeteredLineQuantity(ctx, line, customer, featureMeters) + return e.snapshotMeteredLineQuantity(ctx, line, customer, featureMeters) } -func (s *Service) snapshotLineQuantitiesInParallel(ctx context.Context, customer billing.InvoiceCustomer, lines billing.StandardLines, featureMeters feature.FeatureMeters) error { - workerCount := s.maxParallelQuantitySnapshots +func (e *Engine) snapshotLineQuantitiesInParallel(ctx context.Context, customer billing.InvoiceCustomer, lines billing.StandardLines, featureMeters feature.FeatureMeters) error { + workerCount := e.maxParallelQuantitySnapshots if workerCount <= 0 { workerCount = 1 } @@ -128,7 +178,7 @@ func (s *Service) snapshotLineQuantitiesInParallel(ctx context.Context, customer } }() - err = s.snapshotLineQuantity(ctx, customer, line, featureMeters) + err = e.snapshotLineQuantity(ctx, customer, line, featureMeters) }) } @@ -189,7 +239,7 @@ type featureUsageResponse struct { PreLinePeriodQty alpacadecimal.Decimal } -func (s *Service) getFeatureUsage(ctx context.Context, in getFeatureUsageInput) (*featureUsageResponse, error) { +func (e *Engine) getFeatureUsage(ctx context.Context, in getFeatureUsageInput) (*featureUsageResponse, error) { // Validation if err := in.Validate(); err != nil { return nil, err @@ -206,7 +256,7 @@ func (s *Service) getFeatureUsage(ctx context.Context, in getFeatureUsageInput) // If we are the first line in the split, we don't need to calculate the pre period if lineHierarchy == nil || lineHierarchy.Group.ServicePeriod.From.Equal(in.Line.Period.From) { - meterValues, err := s.streamingConnector.QueryMeter( + meterValues, err := e.streamingConnector.QueryMeter( ctx, in.Line.Namespace, in.Meter, @@ -226,7 +276,7 @@ func (s *Service) getFeatureUsage(ctx context.Context, in getFeatureUsageInput) preLineQuery.From = &lineHierarchy.Group.ServicePeriod.From preLineQuery.To = &in.Line.Period.From - preLineResult, err := s.streamingConnector.QueryMeter( + preLineResult, err := e.streamingConnector.QueryMeter( ctx, in.Line.Namespace, in.Meter, @@ -243,7 +293,7 @@ func (s *Service) getFeatureUsage(ctx context.Context, in getFeatureUsageInput) upToLineEnd.From = &lineHierarchy.Group.ServicePeriod.From upToLineEnd.To = &in.Line.Period.To - upToLineEndResult, err := s.streamingConnector.QueryMeter( + upToLineEndResult, err := e.streamingConnector.QueryMeter( ctx, in.Line.Namespace, in.Meter, diff --git a/openmeter/billing/lineengine/stdinvoice.go b/openmeter/billing/lineengine/stdinvoice.go index ef7f27eb52..4e87f990ee 100644 --- a/openmeter/billing/lineengine/stdinvoice.go +++ b/openmeter/billing/lineengine/stdinvoice.go @@ -10,10 +10,6 @@ import ( "github.com/openmeterio/openmeter/openmeter/billing/service/invoicecalc" ) -type QuantitySnapshotter interface { - SnapshotLineQuantities(ctx context.Context, invoice billing.StandardInvoice, lines billing.StandardLines) error -} - func (e *Engine) BuildStandardInvoiceLines(ctx context.Context, input billing.BuildStandardInvoiceLinesInput) (billing.StandardLines, error) { stdLines, err := e.buildStandardInvoiceLinesWithQuantitySnapshot(ctx, input) if err != nil { @@ -48,7 +44,7 @@ func (e *Engine) buildStandardInvoiceLinesWithQuantitySnapshot(ctx context.Conte return nil, fmt.Errorf("resolving split line group headers: %w", err) } - if err := e.quantitySnapshotter.SnapshotLineQuantities(ctx, input.Invoice, stdLines); err != nil { + if err := e.SnapshotLineQuantities(ctx, input.Invoice, stdLines); err != nil { return nil, fmt.Errorf("snapshotting line quantities: %w", err) } diff --git a/openmeter/billing/service.go b/openmeter/billing/service.go index 0b95b64f25..5636406cf3 100644 --- a/openmeter/billing/service.go +++ b/openmeter/billing/service.go @@ -11,7 +11,6 @@ import ( type Service interface { ProfileService CustomerOverrideService - InvoiceLineService LineEngineService SplitLineGroupService InvoiceService @@ -45,12 +44,6 @@ type CustomerOverrideService interface { ListCustomerOverrides(ctx context.Context, input ListCustomerOverridesInput) (ListCustomerOverridesResult, error) } -type InvoiceLineService interface { - // SnapshotLineQuantity returns an updated line with the quantity snapshoted from meters - // the invoice is used as contextual information to the call. - SnapshotLineQuantity(ctx context.Context, input SnapshotLineQuantityInput) (*StandardLine, error) -} - type LineEngineService interface { RegisterLineEngine(engine LineEngine) error RegisterCreateLineRouter(router CreateLineRouter) error diff --git a/openmeter/billing/service/service.go b/openmeter/billing/service/service.go index d1c57a7a2d..ad8307428d 100644 --- a/openmeter/billing/service/service.go +++ b/openmeter/billing/service/service.go @@ -15,7 +15,6 @@ import ( "github.com/openmeterio/openmeter/openmeter/customer" "github.com/openmeterio/openmeter/openmeter/meter" "github.com/openmeterio/openmeter/openmeter/productcatalog/feature" - "github.com/openmeterio/openmeter/openmeter/streaming" "github.com/openmeterio/openmeter/openmeter/taxcode" "github.com/openmeterio/openmeter/openmeter/watermill/eventbus" "github.com/openmeterio/openmeter/pkg/framework/transaction" @@ -25,43 +24,40 @@ import ( var _ billing.Service = (*Service)(nil) type Service struct { - adapter billing.Adapter - sequenceService sequence.Service - customerService customer.Service - appService app.Service - taxCodeService taxcode.Service - logger *slog.Logger - invoiceCalculator invoicecalc.Calculator - lineEngines *engineRegistry - ratingService rating.Service - featureService feature.FeatureConnector - meterService meter.Service - streamingConnector streaming.Connector + adapter billing.Adapter + sequenceService sequence.Service + customerService customer.Service + appService app.Service + taxCodeService taxcode.Service + logger *slog.Logger + invoiceCalculator invoicecalc.Calculator + lineEngines *engineRegistry + ratingService rating.Service + featureService feature.FeatureConnector + meterService meter.Service publisher eventbus.Publisher - advancementStrategy billing.AdvancementStrategy - fsNamespaceLockdown []string - maxParallelQuantitySnapshots int + advancementStrategy billing.AdvancementStrategy + fsNamespaceLockdown []string standardInvoiceHooks *billing.StandardInvoiceHooks } type Config struct { - Adapter billing.Adapter - SequenceService sequence.Service - CustomerService customer.Service - AppService app.Service - TaxCodeService taxcode.Service - RatingService rating.Service - Logger *slog.Logger - FeatureService feature.FeatureConnector - MeterService meter.Service - StreamingConnector streaming.Connector - Publisher eventbus.Publisher - AdvancementStrategy billing.AdvancementStrategy - FSNamespaceLockdown []string - MaxParallelQuantitySnapshots int + Adapter billing.Adapter + SequenceService sequence.Service + CustomerService customer.Service + AppService app.Service + TaxCodeService taxcode.Service + RatingService rating.Service + LegacyBillingLineEngine *billinglineengine.Engine + Logger *slog.Logger + FeatureService feature.FeatureConnector + MeterService meter.Service + Publisher eventbus.Publisher + AdvancementStrategy billing.AdvancementStrategy + FSNamespaceLockdown []string } func (c Config) Validate() error { @@ -89,6 +85,10 @@ func (c Config) Validate() error { return errors.New("rating service cannot be null") } + if c.LegacyBillingLineEngine == nil { + return errors.New("legacy billing line engine cannot be null") + } + if c.Logger == nil { return errors.New("logger cannot be null") } @@ -101,10 +101,6 @@ func (c Config) Validate() error { return errors.New("meter repo cannot be null") } - if c.StreamingConnector == nil { - return errors.New("streaming connector cannot be null") - } - if c.Publisher == nil { return errors.New("publisher cannot be null") } @@ -113,10 +109,6 @@ func (c Config) Validate() error { return fmt.Errorf("validating advancement strategy: %w", err) } - if c.MaxParallelQuantitySnapshots < 1 { - return errors.New("max parallel snapshots must be greater than 0") - } - return nil } @@ -126,35 +118,24 @@ func New(config Config) (*Service, error) { } svc := &Service{ - adapter: config.Adapter, - sequenceService: config.SequenceService, - customerService: config.CustomerService, - appService: config.AppService, - taxCodeService: config.TaxCodeService, - logger: config.Logger, - ratingService: config.RatingService, - featureService: config.FeatureService, - meterService: config.MeterService, - streamingConnector: config.StreamingConnector, - publisher: config.Publisher, - advancementStrategy: config.AdvancementStrategy, - fsNamespaceLockdown: config.FSNamespaceLockdown, - maxParallelQuantitySnapshots: config.MaxParallelQuantitySnapshots, - invoiceCalculator: invoicecalc.New(), - lineEngines: newEngineRegistry(), - standardInvoiceHooks: models.NewServiceHookRegistry[billing.StandardInvoice](), - } - - invoiceEngine, err := billinglineengine.New(billinglineengine.Config{ - SplitLineGroupAdapter: config.Adapter, - QuantitySnapshotter: svc, - RatingService: config.RatingService, - }) - if err != nil { - return nil, fmt.Errorf("creating invoice engine: %w", err) - } - - if err := svc.RegisterLineEngine(invoiceEngine); err != nil { + adapter: config.Adapter, + sequenceService: config.SequenceService, + customerService: config.CustomerService, + appService: config.AppService, + taxCodeService: config.TaxCodeService, + logger: config.Logger, + ratingService: config.RatingService, + featureService: config.FeatureService, + meterService: config.MeterService, + publisher: config.Publisher, + advancementStrategy: config.AdvancementStrategy, + fsNamespaceLockdown: config.FSNamespaceLockdown, + invoiceCalculator: invoicecalc.New(), + lineEngines: newEngineRegistry(), + standardInvoiceHooks: models.NewServiceHookRegistry[billing.StandardInvoice](), + } + + if err := svc.RegisterLineEngine(config.LegacyBillingLineEngine); err != nil { return nil, fmt.Errorf("registering invoice engine: %w", err) } diff --git a/openmeter/billing/service/stdinvoiceline.go b/openmeter/billing/service/stdinvoiceline.go index b45d467ced..2814ccaa68 100644 --- a/openmeter/billing/service/stdinvoiceline.go +++ b/openmeter/billing/service/stdinvoiceline.go @@ -21,8 +21,6 @@ import ( "github.com/openmeterio/openmeter/pkg/sortx" ) -var _ billing.InvoiceLineService = (*Service)(nil) - // TODO[later]: Move this to gatheringinvoice.go func (s *Service) CreatePendingInvoiceLines(ctx context.Context, input billing.CreatePendingInvoiceLinesInput) (*billing.CreatePendingInvoiceLinesResult, error) { for i := range input.Lines { diff --git a/openmeter/billing/stdinvoiceline.go b/openmeter/billing/stdinvoiceline.go index d6a5422d78..b876a6d625 100644 --- a/openmeter/billing/stdinvoiceline.go +++ b/openmeter/billing/stdinvoiceline.go @@ -1177,20 +1177,3 @@ type GetInvoiceLineInput = LineID type GetInvoiceLineOwnershipAdapterInput = LineID type DeleteInvoiceLineInput = LineID - -type SnapshotLineQuantityInput struct { - Invoice *StandardInvoice - Line *StandardLine -} - -func (i SnapshotLineQuantityInput) Validate() error { - if i.Invoice == nil { - return errors.New("invoice is required") - } - - if i.Line == nil { - return errors.New("line is required") - } - - return nil -} diff --git a/openmeter/billing/worker/subscriptionsync/service/base_test.go b/openmeter/billing/worker/subscriptionsync/service/base_test.go index 9237156d3b..391ed92d1b 100644 --- a/openmeter/billing/worker/subscriptionsync/service/base_test.go +++ b/openmeter/billing/worker/subscriptionsync/service/base_test.go @@ -70,6 +70,7 @@ func (s *SuiteBase) SetupSuite() { service, err := New(Config{ BillingService: s.BillingService, + LegacyBillingLineEngine: s.LegacyBillingLineEngine, Logger: slog.Default(), Tracer: noop.NewTracerProvider().Tracer("test"), SubscriptionSyncAdapter: adapter, @@ -93,6 +94,7 @@ func (s *SuiteBase) setupChargesService(config chargestestutils.Config) { service, err := New(Config{ BillingService: s.BillingService, + LegacyBillingLineEngine: s.LegacyBillingLineEngine, ChargesService: s.Charges, Logger: slog.Default(), Tracer: noop.NewTracerProvider().Tracer("test"), diff --git a/openmeter/billing/worker/subscriptionsync/service/reconciler/invoiceupdater/invoiceupdate.go b/openmeter/billing/worker/subscriptionsync/service/reconciler/invoiceupdater/invoiceupdate.go index cc4aa82fcb..2f5133f440 100644 --- a/openmeter/billing/worker/subscriptionsync/service/reconciler/invoiceupdater/invoiceupdate.go +++ b/openmeter/billing/worker/subscriptionsync/service/reconciler/invoiceupdater/invoiceupdate.go @@ -2,6 +2,7 @@ package invoiceupdater import ( "context" + "errors" "fmt" "log/slog" "strings" @@ -9,6 +10,7 @@ import ( "github.com/samber/lo" "github.com/openmeterio/openmeter/openmeter/billing" + billinglineengine "github.com/openmeterio/openmeter/openmeter/billing/lineengine" "github.com/openmeterio/openmeter/openmeter/customer" "github.com/openmeterio/openmeter/openmeter/streaming" "github.com/openmeterio/openmeter/pkg/clock" @@ -18,16 +20,52 @@ import ( const subscriptionSyncComponentName billing.ComponentName = "subscription-sync" +// TODO: Move invoiceupdater into billing/lineengine: invoice patches are only used by the legacy +// invoicing backend, which is why this package declares the snapshotter dependency directly. +type QuantitySnapshotter interface { + SnapshotLineQuantity(ctx context.Context, input billinglineengine.SnapshotLineQuantityInput) (*billing.StandardLine, error) +} + +type Config struct { + BillingService billing.Service + QuantitySnapshotter QuantitySnapshotter + Logger *slog.Logger +} + +func (c Config) Validate() error { + var errs []error + + if c.BillingService == nil { + errs = append(errs, errors.New("billing service is required")) + } + + if c.QuantitySnapshotter == nil { + errs = append(errs, errors.New("quantity snapshotter is required")) + } + + if c.Logger == nil { + errs = append(errs, errors.New("logger is required")) + } + + return models.NewNillableGenericValidationError(errors.Join(errs...)) +} + type Updater struct { - billingService billing.Service - logger *slog.Logger + billingService billing.Service + quantitySnapshotter QuantitySnapshotter + logger *slog.Logger } -func New(billingService billing.Service, logger *slog.Logger) *Updater { - return &Updater{ - billingService: billingService, - logger: logger, +func New(config Config) (*Updater, error) { + if err := config.Validate(); err != nil { + return nil, err } + + return &Updater{ + billingService: config.BillingService, + quantitySnapshotter: config.QuantitySnapshotter, + logger: config.Logger, + }, nil } func (u *Updater) ApplyPatches(ctx context.Context, customerID customer.CustomerID, patches []Patch) error { @@ -320,7 +358,7 @@ func (u *Updater) updateMutableStandardInvoice(ctx context.Context, invoice bill return fmt.Errorf("line[%s] not found in the invoice, cannot update", targetStandardLine.ID) } - updatedQtyLine, err := u.billingService.SnapshotLineQuantity(ctx, billing.SnapshotLineQuantityInput{ + updatedQtyLine, err := u.quantitySnapshotter.SnapshotLineQuantity(ctx, billinglineengine.SnapshotLineQuantityInput{ Invoice: invoice, Line: &targetStandardLine, }) @@ -465,7 +503,7 @@ func (u *Updater) updateImmutableInvoice(ctx context.Context, invoice billing.St return fmt.Errorf("line[%s] is not a standard line, cannot update: %w", targetState.GetID(), err) } - targetStateWithUpdatedQty, err := u.billingService.SnapshotLineQuantity(ctx, billing.SnapshotLineQuantityInput{ + targetStateWithUpdatedQty, err := u.quantitySnapshotter.SnapshotLineQuantity(ctx, billinglineengine.SnapshotLineQuantityInput{ Invoice: &invoice, Line: &targetStandardLine, }) diff --git a/openmeter/billing/worker/subscriptionsync/service/reconciler/reconciler.go b/openmeter/billing/worker/subscriptionsync/service/reconciler/reconciler.go index 57c8144e33..2201a756ab 100644 --- a/openmeter/billing/worker/subscriptionsync/service/reconciler/reconciler.go +++ b/openmeter/billing/worker/subscriptionsync/service/reconciler/reconciler.go @@ -12,6 +12,7 @@ import ( "github.com/openmeterio/openmeter/openmeter/billing" "github.com/openmeterio/openmeter/openmeter/billing/charges" + billinglineengine "github.com/openmeterio/openmeter/openmeter/billing/lineengine" "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" @@ -29,6 +30,7 @@ type Reconciler interface { type Config struct { BillingService billing.Service + LegacyBillingLineEngine *billinglineengine.Engine ChargesService charges.Service EnableCreditThenInvoice bool Logger *slog.Logger @@ -40,6 +42,10 @@ func (c Config) Validate() error { return fmt.Errorf("billing service is required") } + if c.LegacyBillingLineEngine == nil { + return fmt.Errorf("legacy billing line engine is required") + } + if c.Logger == nil { return fmt.Errorf("logger is required") } @@ -70,10 +76,19 @@ func New(config Config) (*Service, error) { return nil, err } + invoiceUpdater, err := invoiceupdater.New(invoiceupdater.Config{ + BillingService: config.BillingService, + QuantitySnapshotter: config.LegacyBillingLineEngine, + Logger: config.Logger, + }) + if err != nil { + return nil, fmt.Errorf("creating invoice updater: %w", err) + } + return &Service{ billingService: config.BillingService, logger: config.Logger, - invoiceUpdater: invoiceupdater.New(config.BillingService, config.Logger), + invoiceUpdater: invoiceUpdater, chargesService: config.ChargesService, enableCreditThenInvoice: config.EnableCreditThenInvoice && config.ChargesService != nil, featureGate: config.FeatureGate, diff --git a/openmeter/billing/worker/subscriptionsync/service/service.go b/openmeter/billing/worker/subscriptionsync/service/service.go index 19240b4371..45b0c41124 100644 --- a/openmeter/billing/worker/subscriptionsync/service/service.go +++ b/openmeter/billing/worker/subscriptionsync/service/service.go @@ -10,6 +10,7 @@ import ( "github.com/openmeterio/openmeter/openmeter/billing" "github.com/openmeterio/openmeter/openmeter/billing/charges" + billinglineengine "github.com/openmeterio/openmeter/openmeter/billing/lineengine" "github.com/openmeterio/openmeter/openmeter/billing/worker/subscriptionsync" "github.com/openmeterio/openmeter/openmeter/billing/worker/subscriptionsync/service/reconciler" "github.com/openmeterio/openmeter/openmeter/subscription" @@ -26,7 +27,8 @@ type FeatureFlags struct { } type Config struct { - BillingService billing.Service + BillingService billing.Service + LegacyBillingLineEngine *billinglineengine.Engine // ChargesService is required for credit-only sync and charge-based provisioning. ChargesService charges.Service SubscriptionService subscription.Service @@ -43,6 +45,10 @@ func (c Config) Validate() error { return fmt.Errorf("billing service is required") } + if c.LegacyBillingLineEngine == nil { + return fmt.Errorf("legacy billing line engine is required") + } + if c.SubscriptionService == nil { return fmt.Errorf("subscription service is required") } @@ -86,6 +92,7 @@ func New(config Config) (*Service, error) { } reconcilerSvc, err := reconciler.New(reconciler.Config{ BillingService: config.BillingService, + LegacyBillingLineEngine: config.LegacyBillingLineEngine, ChargesService: config.ChargesService, EnableCreditThenInvoice: config.FeatureFlags.EnableCreditThenInvoice, Logger: config.Logger, diff --git a/openmeter/billing/worker/subscriptionsync/service/sync_credittheninvoice_test.go b/openmeter/billing/worker/subscriptionsync/service/sync_credittheninvoice_test.go index c8bfa5aa66..281cd11f9c 100644 --- a/openmeter/billing/worker/subscriptionsync/service/sync_credittheninvoice_test.go +++ b/openmeter/billing/worker/subscriptionsync/service/sync_credittheninvoice_test.go @@ -130,6 +130,7 @@ func (s *CreditThenInvoiceTestSuite) SetupSuite() { service, err := New(Config{ BillingService: s.BillingService, + LegacyBillingLineEngine: s.LegacyBillingLineEngine, ChargesService: s.Charges, Logger: s.Service.logger, Tracer: s.Service.tracer, diff --git a/openmeter/server/server_test.go b/openmeter/server/server_test.go index 32d65e9d0c..058dee4403 100644 --- a/openmeter/server/server_test.go +++ b/openmeter/server/server_test.go @@ -1868,10 +1868,6 @@ func (n NoopBillingService) GetGatheringLinesForSubscription(ctx context.Context return billing.GatheringLines{}, nil } -func (n NoopBillingService) SnapshotLineQuantity(ctx context.Context, input billing.SnapshotLineQuantityInput) (*billing.StandardLine, error) { - return &billing.StandardLine{}, nil -} - // InvoiceSplitLineGroupService methods func (n NoopBillingService) DeleteSplitLineGroup(ctx context.Context, input billing.DeleteSplitLineGroupInput) error { return nil diff --git a/test/app/testenv.go b/test/app/testenv.go index 8f50fb17a1..0efaa38224 100644 --- a/test/app/testenv.go +++ b/test/app/testenv.go @@ -18,6 +18,7 @@ import ( appstripeservice "github.com/openmeterio/openmeter/openmeter/app/stripe/service" "github.com/openmeterio/openmeter/openmeter/billing" billingadapter "github.com/openmeterio/openmeter/openmeter/billing/adapter" + billinglineengine "github.com/openmeterio/openmeter/openmeter/billing/lineengine" billingratingservice "github.com/openmeterio/openmeter/openmeter/billing/rating/service" billingsequenceadapter "github.com/openmeterio/openmeter/openmeter/billing/sequence/adapter" billingsequenceservice "github.com/openmeterio/openmeter/openmeter/billing/sequence/service" @@ -280,20 +281,27 @@ func InitBillingService(t *testing.T, ctx context.Context, in InitBillingService // identically). Lines that carry a unit_config are exercised by the dedicated // unit_config suites. billingRatingService := billingratingservice.New(billingratingservice.Config{UnitConfigEnabled: true}) - - return billingservice.New(billingservice.Config{ - Adapter: billingAdapter, - SequenceService: billingSequenceService, + legacyBillingLineEngine, err := billinglineengine.New(billinglineengine.Config{ + SplitLineGroupAdapter: billingAdapter, RatingService: billingRatingService, - CustomerService: in.CustomerService, - AppService: in.AppService, - Logger: slog.Default(), FeatureService: featureService, - MeterService: meterAdapter, StreamingConnector: mockStreamingConnector, - Publisher: eventbus.NewMock(t), - AdvancementStrategy: billing.ForegroundAdvancementStrategy, MaxParallelQuantitySnapshots: 2, - TaxCodeService: taxCodeService, + }) + require.NoError(t, err) + + return billingservice.New(billingservice.Config{ + Adapter: billingAdapter, + SequenceService: billingSequenceService, + RatingService: billingRatingService, + LegacyBillingLineEngine: legacyBillingLineEngine, + CustomerService: in.CustomerService, + AppService: in.AppService, + Logger: slog.Default(), + FeatureService: featureService, + MeterService: meterAdapter, + Publisher: eventbus.NewMock(t), + AdvancementStrategy: billing.ForegroundAdvancementStrategy, + TaxCodeService: taxCodeService, }) } diff --git a/test/billing/subscription_test.go b/test/billing/subscription_test.go index 26b23bb878..a7f12ef8a6 100644 --- a/test/billing/subscription_test.go +++ b/test/billing/subscription_test.go @@ -50,6 +50,7 @@ func (s *SubscriptionTestSuite) SetupSuite() { service, err := subscriptionsyncservice.New(subscriptionsyncservice.Config{ BillingService: s.BillingService, + LegacyBillingLineEngine: s.LegacyBillingLineEngine, Logger: slog.Default(), Tracer: noop.NewTracerProvider().Tracer("test"), SubscriptionSyncAdapter: subscriptionSyncAdapter, diff --git a/test/billing/suite.go b/test/billing/suite.go index 60cdf2aff1..c6b22ab148 100644 --- a/test/billing/suite.go +++ b/test/billing/suite.go @@ -30,6 +30,7 @@ import ( appservice "github.com/openmeterio/openmeter/openmeter/app/service" "github.com/openmeterio/openmeter/openmeter/billing" billingadapter "github.com/openmeterio/openmeter/openmeter/billing/adapter" + billinglineengine "github.com/openmeterio/openmeter/openmeter/billing/lineengine" "github.com/openmeterio/openmeter/openmeter/billing/models/totals" billingratingservice "github.com/openmeterio/openmeter/openmeter/billing/rating/service" billingsequence "github.com/openmeterio/openmeter/openmeter/billing/sequence" @@ -71,10 +72,11 @@ type BaseSuite struct { TestDB *testutils.TestDB DBClient *db.Client - BillingAdapter billing.Adapter - BillingService billing.Service - SequenceService billingsequence.Service - InvoiceCalculator *invoicecalc.MockableInvoiceCalculator + BillingAdapter billing.Adapter + BillingService billing.Service + LegacyBillingLineEngine *billinglineengine.Engine + SequenceService billingsequence.Service + InvoiceCalculator *invoicecalc.MockableInvoiceCalculator FeatureService feature.FeatureConnector FeatureRepo feature.FeatureRepo @@ -236,20 +238,30 @@ func (s *BaseSuite) setupSuite() { require.NoError(t, err) s.SequenceService = billingSequenceService - billingService, err := billingservice.New(billingservice.Config{ - Adapter: billingAdapter, - SequenceService: billingSequenceService, - RatingService: billingratingservice.New(billingratingservice.Config{UnitConfigEnabled: true}), - CustomerService: s.CustomerService, - AppService: s.AppService, - Logger: slog.Default(), + billingRatingService := billingratingservice.New(billingratingservice.Config{UnitConfigEnabled: true}) + legacyBillingLineEngine, err := billinglineengine.New(billinglineengine.Config{ + SplitLineGroupAdapter: billingAdapter, + RatingService: billingRatingService, FeatureService: s.FeatureService, - MeterService: s.MeterAdapter, StreamingConnector: s.MockStreamingConnector, - Publisher: publisher, - AdvancementStrategy: billing.ForegroundAdvancementStrategy, MaxParallelQuantitySnapshots: 2, - TaxCodeService: taxCodeService, + }) + require.NoError(t, err) + s.LegacyBillingLineEngine = legacyBillingLineEngine + + billingService, err := billingservice.New(billingservice.Config{ + Adapter: billingAdapter, + SequenceService: billingSequenceService, + RatingService: billingRatingService, + LegacyBillingLineEngine: legacyBillingLineEngine, + CustomerService: s.CustomerService, + AppService: s.AppService, + Logger: slog.Default(), + FeatureService: s.FeatureService, + MeterService: s.MeterAdapter, + Publisher: publisher, + AdvancementStrategy: billing.ForegroundAdvancementStrategy, + TaxCodeService: taxCodeService, }) require.NoError(t, err) diff --git a/test/customer/testenv.go b/test/customer/testenv.go index c422fd4fe3..d4f17912a2 100644 --- a/test/customer/testenv.go +++ b/test/customer/testenv.go @@ -18,6 +18,7 @@ import ( appservice "github.com/openmeterio/openmeter/openmeter/app/service" "github.com/openmeterio/openmeter/openmeter/billing" billingadapter "github.com/openmeterio/openmeter/openmeter/billing/adapter" + billinglineengine "github.com/openmeterio/openmeter/openmeter/billing/lineengine" billingratingservice "github.com/openmeterio/openmeter/openmeter/billing/rating/service" billingsequenceadapter "github.com/openmeterio/openmeter/openmeter/billing/sequence/adapter" billingsequenceservice "github.com/openmeterio/openmeter/openmeter/billing/sequence/service" @@ -424,20 +425,31 @@ func NewTestEnv(t *testing.T, ctx context.Context) (TestEnv, error) { return nil, fmt.Errorf("failed to create billing sequence service: %w", err) } - billingService, err := billingservice.New(billingservice.Config{ - Adapter: billingAdapter, - SequenceService: billingSequenceService, - RatingService: billingratingservice.New(billingratingservice.Config{UnitConfigEnabled: true}), - CustomerService: customerService, - AppService: appService, - Logger: logger.WithGroup("billing"), + billingRatingService := billingratingservice.New(billingratingservice.Config{UnitConfigEnabled: true}) + legacyBillingLineEngine, err := billinglineengine.New(billinglineengine.Config{ + SplitLineGroupAdapter: billingAdapter, + RatingService: billingRatingService, FeatureService: entitlementRegistry.Feature, - MeterService: meterService, StreamingConnector: streamingConnector, - Publisher: publisher, - AdvancementStrategy: billing.ForegroundAdvancementStrategy, MaxParallelQuantitySnapshots: 2, - TaxCodeService: taxCodeService, + }) + if err != nil { + return nil, fmt.Errorf("failed to create invoice line engine: %w", err) + } + + billingService, err := billingservice.New(billingservice.Config{ + Adapter: billingAdapter, + SequenceService: billingSequenceService, + RatingService: billingRatingService, + LegacyBillingLineEngine: legacyBillingLineEngine, + CustomerService: customerService, + AppService: appService, + Logger: logger.WithGroup("billing"), + FeatureService: entitlementRegistry.Feature, + MeterService: meterService, + Publisher: publisher, + AdvancementStrategy: billing.ForegroundAdvancementStrategy, + TaxCodeService: taxCodeService, }) if err != nil { return nil, fmt.Errorf("failed to create billing service: %w", err) diff --git a/test/subscription/framework_test.go b/test/subscription/framework_test.go index c701d37f55..a99279bcec 100644 --- a/test/subscription/framework_test.go +++ b/test/subscription/framework_test.go @@ -16,6 +16,7 @@ import ( appservice "github.com/openmeterio/openmeter/openmeter/app/service" "github.com/openmeterio/openmeter/openmeter/billing" billingadapter "github.com/openmeterio/openmeter/openmeter/billing/adapter" + billinglineengine "github.com/openmeterio/openmeter/openmeter/billing/lineengine" billingratingservice "github.com/openmeterio/openmeter/openmeter/billing/rating/service" billingsequenceadapter "github.com/openmeterio/openmeter/openmeter/billing/sequence/adapter" billingsequenceservice "github.com/openmeterio/openmeter/openmeter/billing/sequence/service" @@ -112,20 +113,29 @@ func setup(t *testing.T, _ setupConfig) testDeps { }) require.NoError(t, err) - billingService, err := billingservice.New(billingservice.Config{ - Adapter: billingAdapter, - SequenceService: billingSequenceService, - RatingService: billingratingservice.New(billingratingservice.Config{UnitConfigEnabled: true}), - CustomerService: deps.CustomerService, - AppService: appService, - Logger: slog.Default(), + billingRatingService := billingratingservice.New(billingratingservice.Config{UnitConfigEnabled: true}) + legacyBillingLineEngine, err := billinglineengine.New(billinglineengine.Config{ + SplitLineGroupAdapter: billingAdapter, + RatingService: billingRatingService, FeatureService: deps.FeatureConnector, - MeterService: deps.MeterService, StreamingConnector: deps.MockStreamingConnector, - Publisher: publisher, - AdvancementStrategy: billing.ForegroundAdvancementStrategy, MaxParallelQuantitySnapshots: 2, - TaxCodeService: taxCodeService, + }) + require.NoError(t, err) + + billingService, err := billingservice.New(billingservice.Config{ + Adapter: billingAdapter, + SequenceService: billingSequenceService, + RatingService: billingRatingService, + LegacyBillingLineEngine: legacyBillingLineEngine, + CustomerService: deps.CustomerService, + AppService: appService, + Logger: slog.Default(), + FeatureService: deps.FeatureConnector, + MeterService: deps.MeterService, + Publisher: publisher, + AdvancementStrategy: billing.ForegroundAdvancementStrategy, + TaxCodeService: taxCodeService, }) require.NoError(t, err) @@ -140,6 +150,7 @@ func setup(t *testing.T, _ setupConfig) testDeps { subscriptionSyncService, err := subscriptionsyncservice.New(subscriptionsyncservice.Config{ BillingService: billingService, + LegacyBillingLineEngine: legacyBillingLineEngine, Logger: slog.Default(), Tracer: noop.NewTracerProvider().Tracer("test"), SubscriptionSyncAdapter: subscriptionSyncAdapter,