From 8b49204391079b6bae047a5572ee9c6de6f6b97c Mon Sep 17 00:00:00 2001 From: Peter Turi Date: Sun, 19 Jul 2026 15:35:49 +0200 Subject: [PATCH] feat: move legacy splitline handling to lineengine --- openmeter/billing/adapter.go | 10 - .../service/realizations/credittheninvoice.go | 1 - .../billing/charges/usagebased/rating.go | 13 - .../service/rating/delta/engine_test.go | 7 +- openmeter/billing/gatheringinvoice.go | 38 +- openmeter/billing/invoicelinesplitgroup.go | 457 ------------------ .../legacy/splitlinegroup/adapter.go | 19 + .../adapter/invoicelinesplitgroup.go | 0 .../legacy/splitlinegroup/headers.go | 137 ++++++ .../splitlinegroup/invoicelinesplitgroup.go | 294 +++++++++++ .../invoicelinesplitgroup_test.go | 2 +- .../legacy/splitlinegroup/service.go | 14 + .../service/invoicelinesplitgroup.go | 2 +- openmeter/billing/lineengine/engine.go | 43 +- .../billing/lineengine/quantitysnapshot.go | 42 +- .../billing/lineengine/splitlinegroup.go | 44 +- openmeter/billing/lineengine/stdinvoice.go | 60 ++- openmeter/billing/lineengine/stdline.go | 89 ++++ openmeter/billing/rating/line.go | 4 + openmeter/billing/rating/service.go | 5 + .../billing/rating/service/detailedline.go | 32 +- .../rating/service/mutator/credits_test.go | 24 +- .../service/mutator/forbidunitconfig_test.go | 10 +- .../rating/service/mutator/unitconfig_test.go | 42 +- .../billing/rating/service/rate/types.go | 6 +- .../rating/service/testutil/ubptest.go | 49 +- openmeter/billing/service.go | 8 - openmeter/billing/stdinvoiceedit.go | 20 - openmeter/billing/stdinvoiceline.go | 78 +-- .../service/persistedstate/item.go | 19 +- .../service/persistedstate/loader.go | 50 +- .../invoiceupdater/invoiceupdate.go | 42 +- .../reconciler/invoiceupdater/patch.go | 5 +- .../service/reconciler/patchhelpers.go | 7 +- .../reconciler/patchinvoicelinehierarchy.go | 22 +- 35 files changed, 916 insertions(+), 779 deletions(-) delete mode 100644 openmeter/billing/invoicelinesplitgroup.go create mode 100644 openmeter/billing/invoicing/legacy/splitlinegroup/adapter.go rename openmeter/billing/{ => invoicing/legacy/splitlinegroup}/adapter/invoicelinesplitgroup.go (100%) create mode 100644 openmeter/billing/invoicing/legacy/splitlinegroup/headers.go create mode 100644 openmeter/billing/invoicing/legacy/splitlinegroup/invoicelinesplitgroup.go rename openmeter/billing/{ => invoicing/legacy/splitlinegroup}/invoicelinesplitgroup_test.go (98%) create mode 100644 openmeter/billing/invoicing/legacy/splitlinegroup/service.go rename openmeter/billing/{ => invoicing/legacy/splitlinegroup}/service/invoicelinesplitgroup.go (99%) create mode 100644 openmeter/billing/lineengine/stdline.go diff --git a/openmeter/billing/adapter.go b/openmeter/billing/adapter.go index 8baf3cb7ca..fb5d9f0723 100644 --- a/openmeter/billing/adapter.go +++ b/openmeter/billing/adapter.go @@ -13,7 +13,6 @@ type Adapter interface { ProfileAdapter CustomerOverrideAdapter InvoiceLineAdapter - InvoiceSplitLineGroupAdapter InvoiceAdapter GatheringInvoiceAdapter StandardInvoiceAdapter @@ -91,15 +90,6 @@ type GatheringInvoiceAdapter interface { HardDeleteGatheringInvoiceLines(ctx context.Context, invoiceID InvoiceID, lineIDs []string) error } -type InvoiceSplitLineGroupAdapter interface { - GetSplitLineGroupsForSubscription(ctx context.Context, input GetLinesForSubscriptionInput) ([]SplitLineHierarchy, error) - CreateSplitLineGroup(ctx context.Context, input CreateSplitLineGroupAdapterInput) (SplitLineGroup, error) - UpdateSplitLineGroup(ctx context.Context, input UpdateSplitLineGroupInput) (SplitLineGroup, error) - DeleteSplitLineGroup(ctx context.Context, input DeleteSplitLineGroupInput) error - GetSplitLineGroup(ctx context.Context, input GetSplitLineGroupInput) (SplitLineHierarchy, error) - GetSplitLineGroupHeaders(ctx context.Context, input GetSplitLineGroupHeadersInput) (SplitLineGroupHeaders, error) -} - type InvoiceAppAdapter interface { UpdateInvoiceFields(ctx context.Context, input UpdateInvoiceFieldsInput) error } diff --git a/openmeter/billing/charges/flatfee/service/realizations/credittheninvoice.go b/openmeter/billing/charges/flatfee/service/realizations/credittheninvoice.go index 0c98f6069a..299fec454c 100644 --- a/openmeter/billing/charges/flatfee/service/realizations/credittheninvoice.go +++ b/openmeter/billing/charges/flatfee/service/realizations/credittheninvoice.go @@ -355,7 +355,6 @@ func rateFlatFeeLine(line billing.StandardLine, ratingService billingrating.Serv // split-line metadata must not make the flat pricer skip an otherwise // billable in-advance or in-arrears charge run. ratingLine.SplitLineGroupID = nil - ratingLine.SplitLineHierarchy = nil generatedDetailedLines, err := ratingService.GenerateDetailedLines(ratingLine, billingrating.WithCreditsMutatorDisabled()) if err != nil { diff --git a/openmeter/billing/charges/usagebased/rating.go b/openmeter/billing/charges/usagebased/rating.go index 9b50cbd61a..69dcaa757f 100644 --- a/openmeter/billing/charges/usagebased/rating.go +++ b/openmeter/billing/charges/usagebased/rating.go @@ -66,19 +66,6 @@ func (r RateableIntent) GetStandardLineDiscounts() billing.StandardLineDiscounts return billing.StandardLineDiscounts{} } -func (r RateableIntent) IsProgressivelyBilled() bool { - // A charge is never progressively billed - return false -} - -func (r RateableIntent) GetProgressivelyBilledServicePeriod() (timeutil.ClosedPeriod, error) { - return r.ServicePeriod, nil -} - -func (r RateableIntent) GetPreviouslyBilledAmount() (alpacadecimal.Decimal, error) { - return alpacadecimal.Zero, nil -} - func (r RateableIntent) GetCreditsApplied() billing.CreditsApplied { return r.CreditsApplied } diff --git a/openmeter/billing/charges/usagebased/service/rating/delta/engine_test.go b/openmeter/billing/charges/usagebased/service/rating/delta/engine_test.go index c66fe47125..50331c4393 100644 --- a/openmeter/billing/charges/usagebased/service/rating/delta/engine_test.go +++ b/openmeter/billing/charges/usagebased/service/rating/delta/engine_test.go @@ -1,6 +1,7 @@ package delta import ( + "fmt" "strconv" "testing" "time" @@ -341,7 +342,11 @@ func (s *stubRatingService) ResolveBillablePeriod(in billingrating.ResolveBillab return nil, nil } +func (s *stubRatingService) GenerateProgressiveBilledDetailedLines(in billingrating.ProgressiveBilledLineAccessor, opts ...billingrating.GenerateDetailedLinesOption) (billingrating.GenerateDetailedLinesResult, error) { + return billingrating.GenerateDetailedLinesResult{}, fmt.Errorf("charges rating must always use the non-progressive billed rating engine") +} + func (s *stubRatingService) GenerateDetailedLines(in billingrating.StandardLineAccessor, opts ...billingrating.GenerateDetailedLinesOption) (billingrating.GenerateDetailedLinesResult, error) { s.lastOpts = billingrating.NewGenerateDetailedLinesOptions(opts...) - return s.result, nil + return billingrating.GenerateDetailedLinesResult{}, nil } diff --git a/openmeter/billing/gatheringinvoice.go b/openmeter/billing/gatheringinvoice.go index 33acdb00da..d0fc88247d 100644 --- a/openmeter/billing/gatheringinvoice.go +++ b/openmeter/billing/gatheringinvoice.go @@ -727,8 +727,7 @@ func (i gatheringInvoiceLineGenericWrapper) AsGenericInvoiceLine() GenericInvoic type GatheringLine struct { GatheringLineBase `json:",inline"` - DBState *GatheringLine `json:"-"` - SplitLineHierarchy *SplitLineHierarchy `json:"splitLineHierarchy,omitempty"` + DBState *GatheringLine `json:"-"` } func (g GatheringLine) Clone() (GatheringLine, error) { @@ -805,10 +804,6 @@ func (g GatheringLine) RemoveMetaForCompare() (GatheringLine, error) { return g.WithoutDBState() } -func (g *GatheringLine) SetSplitLineHierarchy(hierarchy *SplitLineHierarchy) { - g.SplitLineHierarchy = hierarchy -} - func (g GatheringLine) AsNewStandardLine(invoiceID string) (*StandardLine, error) { if invoiceID == "" { return nil, errors.New("invoice ID is required") @@ -829,16 +824,6 @@ func (g GatheringLine) AsNewStandardLine(invoiceID string) (*StandardLine, error subscription = g.Subscription.Clone() } - var splitLineHierarchy *SplitLineHierarchy - if g.SplitLineHierarchy != nil { - clonedSHierarchy, err := g.SplitLineHierarchy.Clone() - if err != nil { - return nil, fmt.Errorf("cloning split line hierarchy: %w", err) - } - - splitLineHierarchy = lo.ToPtr(clonedSHierarchy) - } - // Carry the gathering line's unit_config snapshot onto the standard line so the legacy // line-engine path's rating (StandardLine.GetUnitConfig) converts from raw metered units. // Deep-cloned so the standard line owns its own config, matching the charges path. @@ -873,8 +858,6 @@ func (g GatheringLine) AsNewStandardLine(invoiceID string) (*StandardLine, error UnitConfig: unitConfig, }, - SplitLineHierarchy: splitLineHierarchy, - DBState: nil, // We don't want to reuse the state from the gathering line (so let's make it explicit) } @@ -1096,3 +1079,22 @@ type PrepareBillableLinesInput = InvoicePendingLinesInput type PrepareBillableLinesResult struct { LinesByCurrency map[currencyx.Code]GatheringLines } + +type GatheringLineWithInvoiceHeader struct { + Line GatheringLine + Invoice GatheringInvoice +} + +func (l GatheringLineWithInvoiceHeader) Validate() error { + if err := l.Line.Validate(); err != nil { + return fmt.Errorf("line: %w", err) + } + + // This transport only carries the parent invoice entity as a header object, + // so we validate invoice identity here instead of requiring a fully fetched invoice aggregate. + if l.Invoice.ID == "" { + return fmt.Errorf("invoice id is required") + } + + return nil +} diff --git a/openmeter/billing/invoicelinesplitgroup.go b/openmeter/billing/invoicelinesplitgroup.go deleted file mode 100644 index f8683d1b68..0000000000 --- a/openmeter/billing/invoicelinesplitgroup.go +++ /dev/null @@ -1,457 +0,0 @@ -package billing - -import ( - "encoding/json" - "errors" - "fmt" - "time" - - "github.com/alpacahq/alpacadecimal" - "github.com/samber/lo" - - "github.com/openmeterio/openmeter/openmeter/productcatalog" - "github.com/openmeterio/openmeter/pkg/currencyx" - "github.com/openmeterio/openmeter/pkg/models" - "github.com/openmeterio/openmeter/pkg/slicesx" - timeutil "github.com/openmeterio/openmeter/pkg/timeutil" -) - -type SplitLineGroupMutableFields struct { - Name string `json:"name"` - Description *string `json:"description,omitempty"` - Metadata models.Metadata `json:"metadata,omitempty"` - - ServicePeriod timeutil.ClosedPeriod `json:"period"` - - RatecardDiscounts Discounts `json:"ratecardDiscounts"` -} - -func (i SplitLineGroupMutableFields) ValidateForPrice(price *productcatalog.Price) error { - var errs []error - - if i.Name == "" { - errs = append(errs, errors.New("name is required")) - } - - if err := i.ServicePeriod.Validate(); err != nil { - errs = append(errs, err) - } - - if err := i.RatecardDiscounts.ValidateForPrice(price); err != nil { - errs = append(errs, err) - } - - return errors.Join(errs...) -} - -func (i SplitLineGroupMutableFields) Clone() SplitLineGroupMutableFields { - clone := i - clone.RatecardDiscounts = i.RatecardDiscounts.Clone() - - return clone -} - -type SplitLineGroupCreate struct { - Namespace string `json:"namespace"` - - SplitLineGroupMutableFields `json:",inline"` - - Price *productcatalog.Price `json:"price"` - FeatureKey *string `json:"featureKey,omitempty"` - Subscription *SubscriptionReference `json:"subscription,omitempty"` - Currency currencyx.Code `json:"currency"` - UniqueReferenceID *string `json:"childUniqueReferenceId,omitempty"` -} - -func (i SplitLineGroupCreate) Validate() error { - var errs []error - - if i.Namespace == "" { - errs = append(errs, errors.New("namespace is required")) - } - - if err := i.SplitLineGroupMutableFields.ValidateForPrice(i.Price); err != nil { - errs = append(errs, err) - } - - if i.Price == nil { - errs = append(errs, errors.New("price is required")) - } else { - if err := i.Price.Validate(); err != nil { - errs = append(errs, err) - } - } - - if i.Subscription != nil { - if err := i.Subscription.Validate(); err != nil { - errs = append(errs, err) - } - } - - if i.Currency == "" { - errs = append(errs, errors.New("currency is required")) - } - - if i.UniqueReferenceID != nil && *i.UniqueReferenceID == "" { - errs = append(errs, errors.New("unique reference id is required")) - } - - return errors.Join(errs...) -} - -type SplitLineGroupUpdate struct { - models.NamespacedID `json:",inline"` - - SplitLineGroupMutableFields `json:",inline"` -} - -func (i SplitLineGroupUpdate) ValidateWithPrice(price *productcatalog.Price) error { - var errs []error - - if err := i.SplitLineGroupMutableFields.ValidateForPrice(price); err != nil { - errs = append(errs, err) - } - - if err := i.NamespacedID.Validate(); err != nil { - errs = append(errs, err) - } - - return errors.Join(errs...) -} - -type SplitLineGroup struct { - models.ManagedModel `json:",inline"` - models.NamespacedID `json:",inline"` - SplitLineGroupMutableFields `json:",inline"` - - Price *productcatalog.Price `json:"price"` - FeatureKey *string `json:"featureKey,omitempty"` - Subscription *SubscriptionReference `json:"subscription,omitempty"` - Currency currencyx.Code `json:"currency"` - UniqueReferenceID *string `json:"childUniqueReferenceId,omitempty"` -} - -func (i SplitLineGroup) Validate() error { - var errs []error - - if err := i.SplitLineGroupMutableFields.ValidateForPrice(i.Price); err != nil { - errs = append(errs, err) - } - - if i.Price != nil { - if err := i.Price.Validate(); err != nil { - errs = append(errs, err) - } - } - - if i.Currency == "" { - errs = append(errs, errors.New("currency is required")) - } - - return errors.Join(errs...) -} - -func (i SplitLineGroup) ToUpdate() SplitLineGroupUpdate { - return SplitLineGroupUpdate{ - NamespacedID: i.NamespacedID, - SplitLineGroupMutableFields: i.SplitLineGroupMutableFields.Clone(), - } -} - -func (i SplitLineGroup) Clone() SplitLineGroup { - return SplitLineGroup{ - ManagedModel: i.ManagedModel, - NamespacedID: i.NamespacedID, - SplitLineGroupMutableFields: i.SplitLineGroupMutableFields.Clone(), - Price: i.Price, - FeatureKey: i.FeatureKey, - Subscription: i.Subscription, - Currency: i.Currency, - UniqueReferenceID: i.UniqueReferenceID, - } -} - -type GatheringLineWithInvoiceHeader struct { - Line GatheringLine - Invoice GatheringInvoice -} - -func (l GatheringLineWithInvoiceHeader) Validate() error { - if err := l.Line.Validate(); err != nil { - return fmt.Errorf("line: %w", err) - } - - // This transport only carries the parent invoice entity as a header object, - // so we validate invoice identity here instead of requiring a fully fetched invoice aggregate. - if l.Invoice.ID == "" { - return fmt.Errorf("invoice id is required") - } - - return nil -} - -type StandardLineWithInvoiceHeader struct { - Line *StandardLine - Invoice StandardInvoice -} - -func (l StandardLineWithInvoiceHeader) Validate() error { - if l.Line == nil { - return fmt.Errorf("line is required") - } - - if err := l.Line.Validate(); err != nil { - return fmt.Errorf("line: %w", err) - } - - // This transport only carries the parent invoice entity as a header object, - // so we validate invoice identity here instead of requiring a fully fetched invoice aggregate. - if l.Invoice.ID == "" { - return fmt.Errorf("invoice id is required") - } - - return nil -} - -type LineWithInvoiceHeader struct { - Line GenericInvoiceLine - Invoice GenericInvoiceReader -} - -type lineWithInvoiceHeaderSerde[IT StandardInvoice | GatheringInvoice, LT StandardLine | GatheringLine] struct { - Type InvoiceType `json:"type"` - - Line LT `json:"line"` - Invoice IT `json:"invoice"` -} - -func (l *LineWithInvoiceHeader) UnmarshalJSON(data []byte) error { - var genericSerde struct { - Type InvoiceType `json:"type"` - } - - if err := json.Unmarshal(data, &genericSerde); err != nil { - return err - } - - switch genericSerde.Type { - case InvoiceTypeStandard: - unmarshalled := lineWithInvoiceHeaderSerde[StandardInvoice, StandardLine]{} - if err := json.Unmarshal(data, &unmarshalled); err != nil { - return err - } - - l.Line = &standardInvoiceLineGenericWrapper{StandardLine: &unmarshalled.Line} - l.Invoice = &unmarshalled.Invoice - - return nil - case InvoiceTypeGathering: - unmarshalled := lineWithInvoiceHeaderSerde[GatheringInvoice, GatheringLine]{} - if err := json.Unmarshal(data, &unmarshalled); err != nil { - return err - } - - l.Line = &gatheringInvoiceLineGenericWrapper{GatheringLine: unmarshalled.Line} - l.Invoice = &unmarshalled.Invoice - - return nil - default: - return fmt.Errorf("unknown invoice type: %s", genericSerde.Type) - } -} - -func (l LineWithInvoiceHeader) MarshalJSON() ([]byte, error) { - invoice := l.Invoice.AsInvoice() - - invoiceType := invoice.Type() - - switch invoiceType { - case InvoiceTypeStandard: - stdInvoice, err := invoice.AsStandardInvoice() - if err != nil { - return nil, err - } - - stdLine, err := l.Line.AsInvoiceLine().AsStandardLine() - if err != nil { - return nil, err - } - - return json.Marshal(lineWithInvoiceHeaderSerde[StandardInvoice, StandardLine]{ - Type: InvoiceTypeStandard, - Line: stdLine, - Invoice: stdInvoice, - }) - case InvoiceTypeGathering: - gatheringInvoice, err := invoice.AsGatheringInvoice() - if err != nil { - return nil, err - } - - gatheringLine, err := l.Line.AsInvoiceLine().AsGatheringLine() - if err != nil { - return nil, err - } - - return json.Marshal(lineWithInvoiceHeaderSerde[GatheringInvoice, GatheringLine]{ - Type: InvoiceTypeGathering, - Line: gatheringLine, - Invoice: gatheringInvoice, - }) - } - - return nil, fmt.Errorf("unknown invoice type: %s", invoiceType) -} - -func NewLineWithInvoiceHeader[T StandardLineWithInvoiceHeader | GatheringLineWithInvoiceHeader](line T) LineWithInvoiceHeader { - switch v := any(line).(type) { - case StandardLineWithInvoiceHeader: - return LineWithInvoiceHeader{Line: &standardInvoiceLineGenericWrapper{StandardLine: v.Line}, Invoice: v.Invoice} - case GatheringLineWithInvoiceHeader: - return LineWithInvoiceHeader{Line: &gatheringInvoiceLineGenericWrapper{GatheringLine: v.Line}, Invoice: v.Invoice} - } - - return LineWithInvoiceHeader{} -} - -type LinesWithInvoiceHeaders []LineWithInvoiceHeader - -func (i LinesWithInvoiceHeaders) Lines() []GenericInvoiceLine { - return lo.Map(i, func(line LineWithInvoiceHeader, _ int) GenericInvoiceLine { - return line.Line - }) -} - -type SplitLineHierarchy struct { - Group SplitLineGroup - Lines LinesWithInvoiceHeaders -} - -func (h *SplitLineHierarchy) Clone() (SplitLineHierarchy, error) { - lines, err := slicesx.MapWithErr(h.Lines, func(line LineWithInvoiceHeader) (LineWithInvoiceHeader, error) { - clonedLine, err := line.Line.Clone() - if err != nil { - return LineWithInvoiceHeader{}, err - } - - // TODO: We might want to clone the invoice too, but that data is mostly read-only, so it's fine for now. - return LineWithInvoiceHeader{Line: clonedLine, Invoice: line.Invoice}, nil - }) - if err != nil { - return SplitLineHierarchy{}, err - } - - return SplitLineHierarchy{ - Group: h.Group.Clone(), - Lines: lines, - }, nil -} - -type SumNetAmountInput struct { - PeriodEndLTE time.Time - IncludeCharges bool -} - -// SumNetAmount returns the sum of the net amount (pre-tax) of the progressive billed line and its children -// containing the values for all lines whose period's end is <= in.UpTo and are not deleted or not part of -// an invoice that has been deleted. -// As gathering lines do not represent any kind of actual charge, they are not included in the sum. -func (h *SplitLineHierarchy) SumNetAmount(in SumNetAmountInput) (alpacadecimal.Decimal, error) { - netAmount := alpacadecimal.Zero - - err := h.ForEachChild(ForEachChildInput{ - PeriodEndLTE: in.PeriodEndLTE, - Callback: func(child LineWithInvoiceHeader) error { - if child.Invoice.AsInvoice().Type() != InvoiceTypeStandard { - return nil - } - - stdLine, err := child.Line.AsInvoiceLine().AsStandardLine() - if err != nil { - return err - } - - netAmount = netAmount.Add(stdLine.Totals.Amount) - - if in.IncludeCharges { - netAmount = netAmount.Add(stdLine.Totals.ChargesTotal) - } - - return nil - }, - }) - if err != nil { - return alpacadecimal.Zero, err - } - - return netAmount, nil -} - -type ForEachChildInput struct { - PeriodEndLTE time.Time - Callback func(child LineWithInvoiceHeader) error -} - -func (h *SplitLineHierarchy) ForEachChild(in ForEachChildInput) error { - for _, child := range h.Lines { - line := child.Line - - // The line is not in scope - if !in.PeriodEndLTE.IsZero() && line.GetServicePeriod().To.After(in.PeriodEndLTE) { - continue - } - - // The line is deleted - if line.GetDeletedAt() != nil { - continue - } - - // The invoice is deleted - if child.Invoice.GetDeletedAt() != nil { - continue - } - - invoice := child.Invoice.AsInvoice() - if invoice.Type() == InvoiceTypeStandard { - stdInvoice, err := invoice.AsStandardInvoice() - if err != nil { - return err - } - - if stdInvoice.Status == StandardInvoiceStatusDeleted { - continue - } - } - - if err := in.Callback(child); err != nil { - return err - } - } - - return nil -} - -// Adapter -type ( - CreateSplitLineGroupAdapterInput = SplitLineGroupCreate - UpdateSplitLineGroupInput = SplitLineGroupUpdate - DeleteSplitLineGroupInput = models.NamespacedID - GetSplitLineGroupInput = models.NamespacedID -) - -type GetSplitLineGroupHeadersInput struct { - Namespace string - SplitLineGroupIDs []string -} - -type SplitLineGroupHeaders = []SplitLineGroup - -func (i GetSplitLineGroupHeadersInput) Validate() error { - var errs []error - - if i.Namespace == "" { - errs = append(errs, errors.New("namespace is required")) - } - - return errors.Join(errs...) -} diff --git a/openmeter/billing/invoicing/legacy/splitlinegroup/adapter.go b/openmeter/billing/invoicing/legacy/splitlinegroup/adapter.go new file mode 100644 index 0000000000..a996b78bb7 --- /dev/null +++ b/openmeter/billing/invoicing/legacy/splitlinegroup/adapter.go @@ -0,0 +1,19 @@ +package splitlinegroup + +import ( + "context" + + "github.com/openmeterio/openmeter/openmeter/billing" + "github.com/openmeterio/openmeter/pkg/framework/entutils" +) + +type Adapter interface { + GetSplitLineGroupsForSubscription(ctx context.Context, input billing.GetLinesForSubscriptionInput) ([]SplitLineHierarchy, error) + CreateSplitLineGroup(ctx context.Context, input CreateSplitLineGroupAdapterInput) (SplitLineGroup, error) + UpdateSplitLineGroup(ctx context.Context, input UpdateSplitLineGroupInput) (SplitLineGroup, error) + DeleteSplitLineGroup(ctx context.Context, input DeleteSplitLineGroupInput) error + GetSplitLineGroup(ctx context.Context, input GetSplitLineGroupInput) (SplitLineHierarchy, error) + GetSplitLineGroupHeaders(ctx context.Context, input GetSplitLineGroupHeadersInput) (SplitLineGroupHeaders, error) + + entutils.TxCreator +} diff --git a/openmeter/billing/adapter/invoicelinesplitgroup.go b/openmeter/billing/invoicing/legacy/splitlinegroup/adapter/invoicelinesplitgroup.go similarity index 100% rename from openmeter/billing/adapter/invoicelinesplitgroup.go rename to openmeter/billing/invoicing/legacy/splitlinegroup/adapter/invoicelinesplitgroup.go diff --git a/openmeter/billing/invoicing/legacy/splitlinegroup/headers.go b/openmeter/billing/invoicing/legacy/splitlinegroup/headers.go new file mode 100644 index 0000000000..1d61500f65 --- /dev/null +++ b/openmeter/billing/invoicing/legacy/splitlinegroup/headers.go @@ -0,0 +1,137 @@ +package splitlinegroup + +import ( + "fmt" + "time" + + "github.com/openmeterio/openmeter/openmeter/billing" + "github.com/openmeterio/openmeter/openmeter/billing/models/totals" + "github.com/openmeterio/openmeter/pkg/models" + timeutil "github.com/openmeterio/openmeter/pkg/timeutil" +) + +type LineHeaderAccessor interface { + GetDeletedAt() *time.Time + GetServicePeriod() timeutil.ClosedPeriod + GetAnnotations() models.Annotations + GetManagedBy() billing.InvoiceLineManagedBy + GetInvoiceID() models.NamespacedID + GetID() billing.LineID +} + +var _ LineHeaderAccessor = (*StandardLine)(nil) + +type StandardLines []StandardLine + +type StandardLine struct { + ID billing.LineID + DeletedAt *time.Time + Annotations models.Annotations + ManagedBy billing.InvoiceLineManagedBy + + Invoice InvoiceHeader + + ServicePeriod timeutil.ClosedPeriod + Totals totals.Totals + Subscription *billing.SubscriptionReference +} + +func (l StandardLine) GetDeletedAt() *time.Time { + return l.DeletedAt +} + +func (l StandardLine) GetServicePeriod() timeutil.ClosedPeriod { + return l.ServicePeriod +} + +func (l StandardLine) GetAnnotations() models.Annotations { + return l.Annotations +} + +func (l StandardLine) GetManagedBy() billing.InvoiceLineManagedBy { + return l.ManagedBy +} + +func (l StandardLine) GetInvoiceID() models.NamespacedID { + return models.NamespacedID{ + Namespace: l.ID.Namespace, + ID: l.Invoice.ID, + } +} + +func (l StandardLine) GetID() billing.LineID { + return l.ID +} + +func (l StandardLine) Clone() (StandardLine, error) { + var err error + l.Annotations, err = l.Annotations.Clone() + if err != nil { + return StandardLine{}, fmt.Errorf("cloning annotations: %w", err) + } + + return l, nil +} + +var _ LineHeaderAccessor = (*GatheringLine)(nil) + +type GatheringLine struct { + DeletedAt *time.Time + ID billing.LineID + Annotations models.Annotations + ManagedBy billing.InvoiceLineManagedBy + + Invoice InvoiceHeader + + ServicePeriod timeutil.ClosedPeriod + InvoiceAt time.Time + Subscription *billing.SubscriptionReference +} + +func (l GatheringLine) GetDeletedAt() *time.Time { + return l.DeletedAt +} + +func (l GatheringLine) GetServicePeriod() timeutil.ClosedPeriod { + return l.ServicePeriod +} + +func (l GatheringLine) GetAnnotations() models.Annotations { + return l.Annotations +} + +func (l GatheringLine) GetManagedBy() billing.InvoiceLineManagedBy { + return l.ManagedBy +} + +func (l GatheringLine) GetInvoiceID() models.NamespacedID { + return models.NamespacedID{ + Namespace: l.ID.Namespace, + ID: l.Invoice.ID, + } +} + +func (l GatheringLine) GetID() billing.LineID { + return l.ID +} + +func (l *GatheringLine) CloneOrNil() (*GatheringLine, error) { + if l == nil { + return nil, nil + } + + out := *l + annotations, err := out.Annotations.Clone() + if err != nil { + return nil, fmt.Errorf("cloning annotations: %w", err) + } + + out.Annotations = annotations + + return &out, nil +} + +type InvoiceHeader struct { + ID string + DeletedAt *time.Time +} diff --git a/openmeter/billing/invoicing/legacy/splitlinegroup/invoicelinesplitgroup.go b/openmeter/billing/invoicing/legacy/splitlinegroup/invoicelinesplitgroup.go new file mode 100644 index 0000000000..de7ee982bd --- /dev/null +++ b/openmeter/billing/invoicing/legacy/splitlinegroup/invoicelinesplitgroup.go @@ -0,0 +1,294 @@ +package splitlinegroup + +import ( + "errors" + "fmt" + "time" + + "github.com/alpacahq/alpacadecimal" + "github.com/samber/lo" + + "github.com/openmeterio/openmeter/openmeter/billing" + "github.com/openmeterio/openmeter/openmeter/productcatalog" + "github.com/openmeterio/openmeter/pkg/currencyx" + "github.com/openmeterio/openmeter/pkg/models" + timeutil "github.com/openmeterio/openmeter/pkg/timeutil" +) + +type SplitLineGroupMutableFields struct { + Name string `json:"name"` + Description *string `json:"description,omitempty"` + Metadata models.Metadata `json:"metadata,omitempty"` + + ServicePeriod timeutil.ClosedPeriod `json:"period"` + + RatecardDiscounts billing.Discounts `json:"ratecardDiscounts"` +} + +func (i SplitLineGroupMutableFields) ValidateForPrice(price *productcatalog.Price) error { + var errs []error + + if i.Name == "" { + errs = append(errs, errors.New("name is required")) + } + + if err := i.ServicePeriod.Validate(); err != nil { + errs = append(errs, err) + } + + if err := i.RatecardDiscounts.ValidateForPrice(price); err != nil { + errs = append(errs, err) + } + + return errors.Join(errs...) +} + +func (i SplitLineGroupMutableFields) Clone() SplitLineGroupMutableFields { + clone := i + clone.RatecardDiscounts = i.RatecardDiscounts.Clone() + + return clone +} + +type SplitLineGroupCreate struct { + Namespace string `json:"namespace"` + + SplitLineGroupMutableFields `json:",inline"` + + Price *productcatalog.Price `json:"price"` + FeatureKey *string `json:"featureKey,omitempty"` + Subscription *billing.SubscriptionReference `json:"subscription,omitempty"` + Currency currencyx.Code `json:"currency"` + UniqueReferenceID *string `json:"childUniqueReferenceId,omitempty"` +} + +func (i SplitLineGroupCreate) Validate() error { + var errs []error + + if i.Namespace == "" { + errs = append(errs, errors.New("namespace is required")) + } + + if err := i.SplitLineGroupMutableFields.ValidateForPrice(i.Price); err != nil { + errs = append(errs, err) + } + + if i.Price == nil { + errs = append(errs, errors.New("price is required")) + } else { + if err := i.Price.Validate(); err != nil { + errs = append(errs, err) + } + } + + if i.Subscription != nil { + if err := i.Subscription.Validate(); err != nil { + errs = append(errs, err) + } + } + + if i.Currency == "" { + errs = append(errs, errors.New("currency is required")) + } + + if i.UniqueReferenceID != nil && *i.UniqueReferenceID == "" { + errs = append(errs, errors.New("unique reference id is required")) + } + + return errors.Join(errs...) +} + +type SplitLineGroupUpdate struct { + models.NamespacedID `json:",inline"` + + SplitLineGroupMutableFields `json:",inline"` +} + +func (i SplitLineGroupUpdate) ValidateWithPrice(price *productcatalog.Price) error { + var errs []error + + if err := i.SplitLineGroupMutableFields.ValidateForPrice(price); err != nil { + errs = append(errs, err) + } + + if err := i.NamespacedID.Validate(); err != nil { + errs = append(errs, err) + } + + return errors.Join(errs...) +} + +type SplitLineGroup struct { + models.ManagedModel `json:",inline"` + models.NamespacedID `json:",inline"` + SplitLineGroupMutableFields `json:",inline"` + + Price *productcatalog.Price `json:"price"` + FeatureKey *string `json:"featureKey,omitempty"` + Subscription *billing.SubscriptionReference `json:"subscription,omitempty"` + Currency currencyx.Code `json:"currency"` + UniqueReferenceID *string `json:"childUniqueReferenceId,omitempty"` +} + +func (i SplitLineGroup) Validate() error { + var errs []error + + if err := i.SplitLineGroupMutableFields.ValidateForPrice(i.Price); err != nil { + errs = append(errs, err) + } + + if i.Price != nil { + if err := i.Price.Validate(); err != nil { + errs = append(errs, err) + } + } + + if i.Currency == "" { + errs = append(errs, errors.New("currency is required")) + } + + return errors.Join(errs...) +} + +func (i SplitLineGroup) ToUpdate() SplitLineGroupUpdate { + return SplitLineGroupUpdate{ + NamespacedID: i.NamespacedID, + SplitLineGroupMutableFields: i.SplitLineGroupMutableFields.Clone(), + } +} + +func (i SplitLineGroup) Clone() SplitLineGroup { + return SplitLineGroup{ + ManagedModel: i.ManagedModel, + NamespacedID: i.NamespacedID, + SplitLineGroupMutableFields: i.SplitLineGroupMutableFields.Clone(), + Price: i.Price, + FeatureKey: i.FeatureKey, + Subscription: i.Subscription, + Currency: i.Currency, + UniqueReferenceID: i.UniqueReferenceID, + } +} + +type SplitLineHierarchy struct { + Group SplitLineGroup + + StandardLines StandardLines + GatheringLine *GatheringLine +} + +func (h SplitLineHierarchy) Clone() (SplitLineHierarchy, error) { + standardLines, err := lo.MapErr(h.StandardLines, func(line StandardLine, _ int) (StandardLine, error) { + return line.Clone() + }) + if err != nil { + return SplitLineHierarchy{}, fmt.Errorf("cloning standard lines: %w", err) + } + + gatheringLine, err := h.GatheringLine.CloneOrNil() + if err != nil { + return SplitLineHierarchy{}, fmt.Errorf("cloning gathering line: %w", err) + } + + return SplitLineHierarchy{ + Group: h.Group.Clone(), + StandardLines: standardLines, + GatheringLine: gatheringLine, + }, nil +} + +func (h SplitLineHierarchy) Lines() []LineHeaderAccessor { + lines := lo.Map(h.StandardLines, func(line StandardLine, _ int) LineHeaderAccessor { + return line + }) + + if h.GatheringLine != nil { + lines = append(lines, *h.GatheringLine) + } + return lines +} + +type SumNetAmountInput struct { + PeriodEndLTE time.Time + IncludeCharges bool +} + +// SumNetAmount returns the sum of the net amount (pre-tax) of the progressive billed line and its children +// containing the values for all lines whose period's end is <= in.UpTo and are not deleted or not part of +// an invoice that has been deleted. +// As gathering lines do not represent any kind of actual charge, they are not included in the sum. +func (h *SplitLineHierarchy) SumNetAmount(in SumNetAmountInput) (alpacadecimal.Decimal, error) { + netAmount := alpacadecimal.Zero + + err := h.ForEachStandardLine(ForEachStandardLineInput{ + PeriodEndLTE: in.PeriodEndLTE, + Callback: func(line StandardLine) error { + netAmount = netAmount.Add(line.Totals.Amount) + + if in.IncludeCharges { + netAmount = netAmount.Add(line.Totals.ChargesTotal) + } + + return nil + }, + }) + if err != nil { + return alpacadecimal.Zero, err + } + + return netAmount, nil +} + +type ForEachStandardLineInput struct { + PeriodEndLTE time.Time + Callback func(line StandardLine) error +} + +func (h *SplitLineHierarchy) ForEachStandardLine(in ForEachStandardLineInput) error { + for _, line := range h.StandardLines { + // The line is not in scope + if !in.PeriodEndLTE.IsZero() && line.ServicePeriod.To.After(in.PeriodEndLTE) { + continue + } + + if line.DeletedAt != nil { + continue + } + + if line.Invoice.DeletedAt != nil { + continue + } + + if err := in.Callback(line); err != nil { + return err + } + } + + return nil +} + +// Adapter +type ( + // TODO: Remove type aliases + CreateSplitLineGroupAdapterInput = SplitLineGroupCreate + UpdateSplitLineGroupInput = SplitLineGroupUpdate + DeleteSplitLineGroupInput = models.NamespacedID + GetSplitLineGroupInput = models.NamespacedID +) + +type GetSplitLineGroupHeadersInput struct { + Namespace string + SplitLineGroupIDs []string +} + +type SplitLineGroupHeaders = []SplitLineGroup + +func (i GetSplitLineGroupHeadersInput) Validate() error { + var errs []error + + if i.Namespace == "" { + errs = append(errs, errors.New("namespace is required")) + } + + return errors.Join(errs...) +} diff --git a/openmeter/billing/invoicelinesplitgroup_test.go b/openmeter/billing/invoicing/legacy/splitlinegroup/invoicelinesplitgroup_test.go similarity index 98% rename from openmeter/billing/invoicelinesplitgroup_test.go rename to openmeter/billing/invoicing/legacy/splitlinegroup/invoicelinesplitgroup_test.go index d73691c636..fe6c1ad042 100644 --- a/openmeter/billing/invoicelinesplitgroup_test.go +++ b/openmeter/billing/invoicing/legacy/splitlinegroup/invoicelinesplitgroup_test.go @@ -1,4 +1,4 @@ -package billing +package splitlinegroup import ( "encoding/json" diff --git a/openmeter/billing/invoicing/legacy/splitlinegroup/service.go b/openmeter/billing/invoicing/legacy/splitlinegroup/service.go new file mode 100644 index 0000000000..16f64ad50a --- /dev/null +++ b/openmeter/billing/invoicing/legacy/splitlinegroup/service.go @@ -0,0 +1,14 @@ +package splitlinegroup + +import ( + "context" + + "github.com/openmeterio/openmeter/openmeter/billing" +) + +type Service interface { + DeleteSplitLineGroup(ctx context.Context, input DeleteSplitLineGroupInput) error + UpdateSplitLineGroup(ctx context.Context, input UpdateSplitLineGroupInput) (SplitLineGroup, error) + // GetSplitLineGroupsForSubscription returns the active split-line hierarchies required for subscription sync. + GetSplitLineGroupsForSubscription(ctx context.Context, input billing.GetLinesForSubscriptionInput) ([]SplitLineHierarchy, error) +} diff --git a/openmeter/billing/service/invoicelinesplitgroup.go b/openmeter/billing/invoicing/legacy/splitlinegroup/service/invoicelinesplitgroup.go similarity index 99% rename from openmeter/billing/service/invoicelinesplitgroup.go rename to openmeter/billing/invoicing/legacy/splitlinegroup/service/invoicelinesplitgroup.go index 465481492a..a87eff0287 100644 --- a/openmeter/billing/service/invoicelinesplitgroup.go +++ b/openmeter/billing/invoicing/legacy/splitlinegroup/service/invoicelinesplitgroup.go @@ -1,4 +1,4 @@ -package billingservice +package service import ( "context" diff --git a/openmeter/billing/lineengine/engine.go b/openmeter/billing/lineengine/engine.go index c7d75319a7..0a157494b4 100644 --- a/openmeter/billing/lineengine/engine.go +++ b/openmeter/billing/lineengine/engine.go @@ -16,10 +16,7 @@ import ( "github.com/openmeterio/openmeter/pkg/slicesx" ) -var ( - _ billing.LineEngine = (*Engine)(nil) - _ billing.LineCalculator = (*Engine)(nil) -) +var _ billing.LineEngine = (*Engine)(nil) type Config struct { SplitLineGroupAdapter SplitLineGroupAdapter @@ -95,7 +92,12 @@ func (e *Engine) OnCollectionCompleted(ctx context.Context, input billing.OnColl return input.Lines, nil } - if err := e.SnapshotLineQuantities(ctx, input.Invoice, input.Lines); err != nil { + linesWithSplitLineHierarchy, err := e.ResolveSplitLineGroupHeaders(ctx, input.Invoice.Namespace, input.Lines) + if err != nil { + return nil, fmt.Errorf("resolving split line group headers: %w", err) + } + + if err := e.SnapshotLineQuantities(ctx, input.Invoice, linesWithSplitLineHierarchy); err != nil { if _, isInvalidDatabaseState := lo.ErrorsAs[*billing.ErrSnapshotInvalidDatabaseState](err); isInvalidDatabaseState { return nil, billing.ValidationIssue{ Severity: billing.ValidationIssueSeverityCritical, @@ -108,7 +110,15 @@ func (e *Engine) OnCollectionCompleted(ctx context.Context, input billing.OnColl return nil, fmt.Errorf("snapshotting lines: %w", err) } - return input.Lines, nil + calculatedLines, err := e.calculateLines(calculateLinesInput{ + Invoice: input.Invoice, + Lines: linesWithSplitLineHierarchy, + }) + if err != nil { + return nil, fmt.Errorf("calculating lines: %w", err) + } + + return calculatedLines.AsStandardLines(), nil } func (e *Engine) ValidateMutableInvoiceLineEditViaAPI(_ context.Context, input billing.OnMutableInvoiceUpdateInput) error { @@ -190,11 +200,28 @@ func (e *Engine) snapshotManualStandardLineOverrideIfNeeded(ctx context.Context, return nil, fmt.Errorf("getting standard line: %w", err) } - if err := e.SnapshotLineQuantities(ctx, standardInvoice, billing.StandardLines{&standardLine}); err != nil { + linesWithSplitLineHierarchy, err := e.ResolveSplitLineGroupHeaders(ctx, invoice.GetInvoiceID().Namespace, billing.StandardLines{&standardLine}) + if err != nil { + return nil, fmt.Errorf("resolving split line group headers: %w", err) + } + + if err := e.SnapshotLineQuantities(ctx, standardInvoice, linesWithSplitLineHierarchy); err != nil { return nil, fmt.Errorf("snapshotting line quantity: %w", err) } - return standardLine.AsGenericLine(), nil + calculatedLines, err := e.calculateLines(calculateLinesInput{ + Invoice: standardInvoice, + Lines: linesWithSplitLineHierarchy, + }) + if err != nil { + return nil, fmt.Errorf("calculating lines: %w", err) + } + + if len(calculatedLines) != 1 { + return nil, fmt.Errorf("expected 1 line, got %d [line_id=%s]", len(calculatedLines), standardLine.GetID()) + } + + return calculatedLines[0].AsGenericLine(), nil } func validateLegacyLineOverride(override billing.InvoiceLineOverride) error { diff --git a/openmeter/billing/lineengine/quantitysnapshot.go b/openmeter/billing/lineengine/quantitysnapshot.go index 107f985de6..276a5dd316 100644 --- a/openmeter/billing/lineengine/quantitysnapshot.go +++ b/openmeter/billing/lineengine/quantitysnapshot.go @@ -28,6 +28,7 @@ func (i SnapshotLineQuantityInput) Validate() 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")) } @@ -35,6 +36,9 @@ func (i SnapshotLineQuantityInput) Validate() error { return errors.Join(errs...) } +// SnapshotLineQuantity snapshots the quantity of a standard line, this is an external API for invoice updater as needed. +// This is only needed for the invoiceupdater to try to minimize unnecessary immutable invoice updates, but generally should be an internal affair +// of the lineengine itself. func (e *Engine) SnapshotLineQuantity(ctx context.Context, input SnapshotLineQuantityInput) (*billing.StandardLine, error) { if err := input.Validate(); err != nil { return nil, billing.ValidationError{ @@ -42,20 +46,29 @@ func (e *Engine) SnapshotLineQuantity(ctx context.Context, input SnapshotLineQua } } - featureMeters, err := e.resolveFeatureMeters(ctx, input.Invoice.Namespace, billing.StandardLines{input.Line}) + linesWithHierarchy, err := e.ResolveSplitLineGroupHeaders(ctx, input.Invoice.Namespace, billing.StandardLines{input.Line}) + if err != nil { + return nil, fmt.Errorf("resolving split line group headers: %w", err) + } + + featureMeters, err := e.resolveFeatureMeters(ctx, input.Invoice.Namespace, linesWithHierarchy) if err != nil { return nil, fmt.Errorf("line[%s]: %w", input.Line.ID, err) } - err = e.snapshotLineQuantity(ctx, input.Invoice.Customer, input.Line, featureMeters) + if len(linesWithHierarchy) != 1 { + return nil, fmt.Errorf("expected 1 line with hierarchy, got %d [line_id: %s]", len(linesWithHierarchy), input.Line.ID) + } + + err = e.snapshotLineQuantity(ctx, input.Invoice.Customer, linesWithHierarchy[0], featureMeters) if err != nil { return nil, err } - return input.Line, nil + return linesWithHierarchy[0].StandardLine, nil } -func (e *Engine) SnapshotLineQuantities(ctx context.Context, invoice billing.StandardInvoice, lines billing.StandardLines) error { +func (e *Engine) SnapshotLineQuantities(ctx context.Context, invoice billing.StandardInvoice, lines StandardLinesWithSplitLineHierarchy) error { featureMeters, err := e.resolveFeatureMeters(ctx, invoice.Namespace, lines) if err != nil { return fmt.Errorf("resolving feature meters: %w", err) @@ -68,7 +81,7 @@ func (e *Engine) SnapshotLineQuantities(ctx context.Context, invoice billing.Sta return nil } -func (e *Engine) resolveFeatureMeters(ctx context.Context, namespace string, lines billing.StandardLines) (feature.FeatureMeters, error) { +func (e *Engine) resolveFeatureMeters(ctx context.Context, namespace string, lines StandardLinesWithSplitLineHierarchy) (feature.FeatureMeters, error) { keys, err := lines.GetReferencedFeatureKeys() if err != nil { return nil, fmt.Errorf("getting referenced feature keys: %w", err) @@ -100,7 +113,7 @@ func (w featureMetersErrorWrapper) Get(featureKey string, requireMeter bool) (fe return featureMeter, nil } -func (e *Engine) snapshotMeteredLineQuantity(ctx context.Context, line *billing.StandardLine, customer billing.InvoiceCustomer, featureMeters feature.FeatureMeters) error { +func (e *Engine) snapshotMeteredLineQuantity(ctx context.Context, line StandardLineWithSplitLineHierarchy, customer billing.InvoiceCustomer, featureMeters feature.FeatureMeters) error { featureMeter, err := featureMeters.Get(line.UsageBased.FeatureKey, true) if err != nil { return err @@ -126,7 +139,7 @@ func (e *Engine) snapshotMeteredLineQuantity(ctx context.Context, line *billing. return nil } -func (e *Engine) snapshotFlatPriceLineQuantity(_ context.Context, line *billing.StandardLine) error { +func (e *Engine) snapshotFlatPriceLineQuantity(_ context.Context, line StandardLineWithSplitLineHierarchy) error { line.UsageBased.MeteredQuantity = lo.ToPtr(alpacadecimal.NewFromInt(1)) line.UsageBased.Quantity = lo.ToPtr(alpacadecimal.NewFromInt(1)) line.UsageBased.PreLinePeriodQuantity = lo.ToPtr(alpacadecimal.Zero) @@ -134,7 +147,7 @@ func (e *Engine) snapshotFlatPriceLineQuantity(_ context.Context, line *billing. return nil } -func (e *Engine) 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 StandardLineWithSplitLineHierarchy, featureMeters feature.FeatureMeters) error { if !line.DependsOnMeteredQuantity() { return e.snapshotFlatPriceLineQuantity(ctx, line) } @@ -142,7 +155,7 @@ func (e *Engine) snapshotLineQuantity(ctx context.Context, customer billing.Invo return e.snapshotMeteredLineQuantity(ctx, line, customer, featureMeters) } -func (e *Engine) snapshotLineQuantitiesInParallel(ctx context.Context, customer billing.InvoiceCustomer, lines billing.StandardLines, featureMeters feature.FeatureMeters) error { +func (e *Engine) snapshotLineQuantitiesInParallel(ctx context.Context, customer billing.InvoiceCustomer, lines StandardLinesWithSplitLineHierarchy, featureMeters feature.FeatureMeters) error { workerCount := e.maxParallelQuantitySnapshots if workerCount <= 0 { workerCount = 1 @@ -198,14 +211,14 @@ func (e *Engine) snapshotLineQuantitiesInParallel(ctx context.Context, customer } type getFeatureUsageInput struct { - Line *billing.StandardLine + Line StandardLineWithSplitLineHierarchy Meter meter.Meter Feature feature.Feature Customer billing.InvoiceCustomer } func (i getFeatureUsageInput) Validate() error { - if i.Line == nil { + if i.Line.StandardLine == nil { return fmt.Errorf("line is required") } @@ -220,10 +233,11 @@ func (i getFeatureUsageInput) Validate() error { // TODO[OM-160]: We need to have this check to make sure that usage discounts are properly accounted for // but we seem to have a bug in syncing progressively billed lines, so let's address this as a separate pr. + // TODO[BeforeMerge]: Make sure we have a test for this. - // if i.Line.SplitLineGroupID != nil && i.Line.SplitLineHierarchy == nil { - // return fmt.Errorf("split line group id is set but split line hierarchy is not expanded") - // } + if i.Line.SplitLineGroupID != nil && i.Line.SplitLineHierarchy == nil { + return fmt.Errorf("split line group id is set but split line hierarchy is not expanded") + } if err := i.Customer.Validate(); err != nil { return fmt.Errorf("customer: %w", err) diff --git a/openmeter/billing/lineengine/splitlinegroup.go b/openmeter/billing/lineengine/splitlinegroup.go index cc4488afe4..e82550bc50 100644 --- a/openmeter/billing/lineengine/splitlinegroup.go +++ b/openmeter/billing/lineengine/splitlinegroup.go @@ -7,15 +7,17 @@ import ( "github.com/samber/lo" "github.com/openmeterio/openmeter/openmeter/billing" + "github.com/openmeterio/openmeter/openmeter/billing/invoicing/legacy/splitlinegroup" "github.com/openmeterio/openmeter/openmeter/productcatalog" "github.com/openmeterio/openmeter/openmeter/streaming" "github.com/openmeterio/openmeter/pkg/clock" "github.com/openmeterio/openmeter/pkg/timeutil" ) +// TODO: Remove this and use the splitline adapter type SplitLineGroupAdapter interface { - CreateSplitLineGroup(ctx context.Context, input billing.CreateSplitLineGroupAdapterInput) (billing.SplitLineGroup, error) - GetSplitLineGroupHeaders(ctx context.Context, input billing.GetSplitLineGroupHeadersInput) (billing.SplitLineGroupHeaders, error) + CreateSplitLineGroup(ctx context.Context, input splitlinegroup.CreateSplitLineGroupAdapterInput) (splitlinegroup.SplitLineGroup, error) + GetSplitLineGroupHeaders(ctx context.Context, input splitlinegroup.GetSplitLineGroupHeadersInput) (splitlinegroup.SplitLineGroupHeaders, error) } func (e *Engine) SplitGatheringLine(ctx context.Context, in billing.SplitGatheringLineInput) (billing.SplitGatheringLineResult, error) { @@ -33,9 +35,9 @@ func (e *Engine) SplitGatheringLine(ctx context.Context, in billing.SplitGatheri var splitLineGroupID string if line.SplitLineGroupID == nil { - splitLineGroup, err := e.adapter.CreateSplitLineGroup(ctx, billing.CreateSplitLineGroupAdapterInput{ + splitLineGroup, err := e.adapter.CreateSplitLineGroup(ctx, splitlinegroup.CreateSplitLineGroupAdapterInput{ Namespace: line.Namespace, - SplitLineGroupMutableFields: billing.SplitLineGroupMutableFields{ + SplitLineGroupMutableFields: splitlinegroup.SplitLineGroupMutableFields{ Name: line.Name, Description: line.Description, ServicePeriod: timeutil.ClosedPeriod{ @@ -113,7 +115,7 @@ func (e *Engine) SplitGatheringLine(ctx context.Context, in billing.SplitGatheri }, nil } -func (e *Engine) ResolveSplitLineGroupHeaders(ctx context.Context, ns string, lines billing.StandardLines) error { +func (e *Engine) ResolveSplitLineGroupHeaders(ctx context.Context, ns string, lines billing.StandardLines) ([]StandardLineWithSplitLineHierarchy, error) { splitLineGroupIDs := lo.Uniq( lo.Filter( lo.Map(lines, func(line *billing.StandardLine, _ int) string { return lo.FromPtr(line.SplitLineGroupID) }), @@ -122,37 +124,39 @@ func (e *Engine) ResolveSplitLineGroupHeaders(ctx context.Context, ns string, li ) if len(splitLineGroupIDs) == 0 { - return nil + return lo.Map(lines, func(line *billing.StandardLine, _ int) StandardLineWithSplitLineHierarchy { + return StandardLineWithSplitLineHierarchy{StandardLine: line} + }), nil } - splitLineGroupHeaders, err := e.adapter.GetSplitLineGroupHeaders(ctx, billing.GetSplitLineGroupHeadersInput{ + splitLineGroupHeaders, err := e.adapter.GetSplitLineGroupHeaders(ctx, splitlinegroup.GetSplitLineGroupHeadersInput{ Namespace: ns, SplitLineGroupIDs: splitLineGroupIDs, }) if err != nil { - return fmt.Errorf("getting split line group headers: %w", err) + return nil, fmt.Errorf("getting split line group headers: %w", err) } - splitLineGroupHeadersByID := lo.SliceToMap(splitLineGroupHeaders, func(header billing.SplitLineGroup) (string, billing.SplitLineGroup) { + splitLineGroupHeadersByID := lo.SliceToMap(splitLineGroupHeaders, func(header splitlinegroup.SplitLineGroup) (string, splitlinegroup.SplitLineGroup) { return header.ID, header }) - for idx := range lines { - if lines[idx].SplitLineGroupID == nil { - continue + return lo.MapErr(lines, func(line *billing.StandardLine, _ int) (StandardLineWithSplitLineHierarchy, error) { + if line.SplitLineGroupID == nil { + return StandardLineWithSplitLineHierarchy{StandardLine: line}, nil } - splitLineGroupHeader, ok := splitLineGroupHeadersByID[lo.FromPtr(lines[idx].SplitLineGroupID)] + splitLineGroupHeader, ok := splitLineGroupHeadersByID[lo.FromPtr(line.SplitLineGroupID)] if !ok { - return fmt.Errorf("split line group header not found for line[%s]: id[%s]", lines[idx].ID, lo.FromPtr(lines[idx].SplitLineGroupID)) + return StandardLineWithSplitLineHierarchy{StandardLine: line}, fmt.Errorf("split line group header not found for line[%s]: id[%s]", line.ID, lo.FromPtr(line.SplitLineGroupID)) } - lines[idx].SplitLineHierarchy = &billing.SplitLineHierarchy{ - Group: splitLineGroupHeader, - } - } - - return nil + return StandardLineWithSplitLineHierarchy{ + StandardLine: line, SplitLineHierarchy: &splitlinegroup.SplitLineHierarchy{ + Group: splitLineGroupHeader, + }, + }, nil + }) } func isPeriodEmptyConsideringTruncations(line billing.GatheringLine) (bool, error) { diff --git a/openmeter/billing/lineengine/stdinvoice.go b/openmeter/billing/lineengine/stdinvoice.go index 4e87f990ee..691e9e5c5b 100644 --- a/openmeter/billing/lineengine/stdinvoice.go +++ b/openmeter/billing/lineengine/stdinvoice.go @@ -2,6 +2,7 @@ package lineengine import ( "context" + "errors" "fmt" "github.com/samber/lo" @@ -11,15 +12,7 @@ import ( ) func (e *Engine) BuildStandardInvoiceLines(ctx context.Context, input billing.BuildStandardInvoiceLinesInput) (billing.StandardLines, error) { - stdLines, err := e.buildStandardInvoiceLinesWithQuantitySnapshot(ctx, input) - if err != nil { - return nil, err - } - - return e.CalculateLines(billing.CalculateLinesInput{ - Invoice: input.Invoice, - Lines: stdLines, - }) + return e.buildStandardInvoiceLinesWithQuantitySnapshot(ctx, input) } func (e *Engine) BuildStandardLinesForGatheringPreview(ctx context.Context, input billing.BuildStandardInvoiceLinesInput) (billing.StandardLines, error) { @@ -40,33 +33,62 @@ func (e *Engine) buildStandardInvoiceLinesWithQuantitySnapshot(ctx context.Conte return nil, err } - if err := e.ResolveSplitLineGroupHeaders(ctx, input.Invoice.Namespace, stdLines); err != nil { + linesWithSplitLineHierarchy, err := e.ResolveSplitLineGroupHeaders(ctx, input.Invoice.Namespace, stdLines) + if err != nil { return nil, fmt.Errorf("resolving split line group headers: %w", err) } - if err := e.SnapshotLineQuantities(ctx, input.Invoice, stdLines); err != nil { + if err := e.SnapshotLineQuantities(ctx, input.Invoice, linesWithSplitLineHierarchy); err != nil { return nil, fmt.Errorf("snapshotting line quantities: %w", err) } - return stdLines, nil + calculatedLines, err := e.calculateLines(calculateLinesInput{ + Invoice: input.Invoice, + Lines: linesWithSplitLineHierarchy, + }) + if err != nil { + return nil, err + } + + return calculatedLines.AsStandardLines(), nil } -func (e *Engine) CalculateLines(input billing.CalculateLinesInput) (billing.StandardLines, error) { - if input.Invoice.ID == "" { - return nil, fmt.Errorf("invoice id is required") +type calculateLinesInput struct { + Invoice billing.StandardInvoice + Lines StandardLinesWithSplitLineHierarchy +} + +func (i calculateLinesInput) Validate() error { + var errs []error + + if i.Invoice.ID == "" { + errs = append(errs, fmt.Errorf("invoice id is required")) } - if len(input.Lines) == 0 { - return nil, fmt.Errorf("lines are required") + if len(i.Lines) == 0 { + errs = append(errs, fmt.Errorf("lines are required")) + } + + if err := i.Lines.Validate(); err != nil { + errs = append(errs, fmt.Errorf("lines: %w", err)) + } + + return errors.Join(errs...) +} + +// TODO: Do not implement the calculator here, so that we don't need to pass the splitline groups to the rating service +func (e *Engine) calculateLines(input calculateLinesInput) (StandardLinesWithSplitLineHierarchy, error) { + if err := input.Validate(); err != nil { + return nil, fmt.Errorf("validating input: %w", err) } for _, stdLine := range input.Lines { - generatedDetailedLines, err := e.ratingService.GenerateDetailedLines(stdLine) + generatedDetailedLines, err := e.ratingService.GenerateProgressiveBilledDetailedLines(stdLine) if err != nil { return nil, fmt.Errorf("generating detailed lines for line[%s]: %w", stdLine.ID, err) } - if err := invoicecalc.MergeGeneratedDetailedLines(stdLine, generatedDetailedLines); err != nil { + if err := invoicecalc.MergeGeneratedDetailedLines(stdLine.StandardLine, generatedDetailedLines); err != nil { return nil, fmt.Errorf("merging generated detailed lines for line[%s]: %w", stdLine.ID, err) } diff --git a/openmeter/billing/lineengine/stdline.go b/openmeter/billing/lineengine/stdline.go new file mode 100644 index 0000000000..0427e14638 --- /dev/null +++ b/openmeter/billing/lineengine/stdline.go @@ -0,0 +1,89 @@ +package lineengine + +import ( + "errors" + "fmt" + + "github.com/alpacahq/alpacadecimal" + "github.com/openmeterio/openmeter/openmeter/billing" + "github.com/openmeterio/openmeter/openmeter/billing/invoicing/legacy/splitlinegroup" + "github.com/openmeterio/openmeter/pkg/models" + "github.com/openmeterio/openmeter/pkg/timeutil" + "github.com/samber/lo" +) + +type StandardLineWithSplitLineHierarchy struct { + *billing.StandardLine + SplitLineHierarchy *splitlinegroup.SplitLineHierarchy +} + +func (i StandardLineWithSplitLineHierarchy) Validate() error { + if i.StandardLine == nil { + return models.NewNillableGenericValidationError(errors.New("standard line is required")) + } + + var errs []error + + if err := i.StandardLine.Validate(); err != nil { + errs = append(errs, err) + } + + if i.StandardLine.SplitLineGroupID != nil { + if i.SplitLineHierarchy == nil { + errs = append(errs, errors.New("split line hierarchy is required")) + } + } + + return errors.Join(errs...) +} + +func (l StandardLineWithSplitLineHierarchy) IsProgressivelyBilled() bool { + return l.SplitLineGroupID != nil +} + +func (i StandardLineWithSplitLineHierarchy) GetProgressivelyBilledServicePeriod() (timeutil.ClosedPeriod, error) { + if i.SplitLineGroupID == nil { + return timeutil.ClosedPeriod{ + From: i.Period.From, + To: i.Period.To, + }, nil + } + + if i.SplitLineHierarchy == nil { + return timeutil.ClosedPeriod{}, errors.New("split line hierarchy is required") + } + + return i.SplitLineHierarchy.Group.ServicePeriod, nil +} + +func (i StandardLineWithSplitLineHierarchy) GetPreviouslyBilledAmount() (alpacadecimal.Decimal, error) { + if i.SplitLineGroupID == nil { + return alpacadecimal.Zero, nil + } + + if i.SplitLineHierarchy == nil { + return alpacadecimal.Zero, fmt.Errorf("line[%s] does not have a progressive line hierarchy, but is a progressive billed line", i.ID) + } + + return i.SplitLineHierarchy.SumNetAmount(splitlinegroup.SumNetAmountInput{ + PeriodEndLTE: i.Period.From, + }) +} + +type StandardLinesWithSplitLineHierarchy []StandardLineWithSplitLineHierarchy + +func (i StandardLinesWithSplitLineHierarchy) Validate() error { + return errors.Join(lo.Map(i, func(line StandardLineWithSplitLineHierarchy, _ int) error { + return line.Validate() + })...) +} + +func (i StandardLinesWithSplitLineHierarchy) AsStandardLines() billing.StandardLines { + return lo.Map(i, func(line StandardLineWithSplitLineHierarchy, _ int) *billing.StandardLine { + return line.StandardLine + }) +} + +func (i StandardLinesWithSplitLineHierarchy) GetReferencedFeatureKeys() ([]string, error) { + return i.AsStandardLines().GetReferencedFeatureKeys() +} diff --git a/openmeter/billing/rating/line.go b/openmeter/billing/rating/line.go index de7caf0973..2d2cf75370 100644 --- a/openmeter/billing/rating/line.go +++ b/openmeter/billing/rating/line.go @@ -44,6 +44,10 @@ type StandardLineAccessor interface { GetUnitConfig() *productcatalog.UnitConfig // GetStandardLineDiscounts returns the standard line discounts for the line GetStandardLineDiscounts() billing.StandardLineDiscounts +} + +type ProgressiveBilledLineAccessor interface { + StandardLineAccessor // Progressive billing related information // IsProgressivelyBilled returns true if the line is progressively billed diff --git a/openmeter/billing/rating/service.go b/openmeter/billing/rating/service.go index 562a3054d5..cbaa3a8fe2 100644 --- a/openmeter/billing/rating/service.go +++ b/openmeter/billing/rating/service.go @@ -14,6 +14,11 @@ import ( type Service interface { ResolveBillablePeriod(in ResolveBillablePeriodInput) (*timeutil.ClosedPeriod, error) + // GenerateProgressiveBilledDetailedLines is a specialized method for generating detailed lines for progressively billed lines. Charges + // never uses this, as it's rating logic does not rely on interval slicing. Once legacy billing is removed, this must be removed, making + // rating code easier to understand/less complex. + GenerateProgressiveBilledDetailedLines(in ProgressiveBilledLineAccessor, opts ...GenerateDetailedLinesOption) (GenerateDetailedLinesResult, error) + // GenerateDetailedLines is the general method for generating detailed lines for standard lines. Progressive billing is not supported. GenerateDetailedLines(in StandardLineAccessor, opts ...GenerateDetailedLinesOption) (GenerateDetailedLinesResult, error) } diff --git a/openmeter/billing/rating/service/detailedline.go b/openmeter/billing/rating/service/detailedline.go index fd08639b9d..d5008735d6 100644 --- a/openmeter/billing/rating/service/detailedline.go +++ b/openmeter/billing/rating/service/detailedline.go @@ -12,10 +12,11 @@ import ( "github.com/openmeterio/openmeter/openmeter/billing/rating/service/rate" "github.com/openmeterio/openmeter/openmeter/productcatalog" "github.com/openmeterio/openmeter/pkg/currencyx" + "github.com/openmeterio/openmeter/pkg/timeutil" ) // ValidateStandardLine validates the standard line and returns an error if the line is invalid/inconsistent -func validateStandardLine(in rating.StandardLineAccessor) error { +func validateStandardLine(in rating.ProgressiveBilledLineAccessor) error { if in == nil { return fmt.Errorf("line is nil") } @@ -38,7 +39,7 @@ func validateStandardLine(in rating.StandardLineAccessor) error { return nil } -func (s *service) GenerateDetailedLines(in rating.StandardLineAccessor, opts ...rating.GenerateDetailedLinesOption) (rating.GenerateDetailedLinesResult, error) { +func (s *service) GenerateProgressiveBilledDetailedLines(in rating.ProgressiveBilledLineAccessor, opts ...rating.GenerateDetailedLinesOption) (rating.GenerateDetailedLinesResult, error) { if err := validateStandardLine(in); err != nil { return rating.GenerateDetailedLinesResult{}, fmt.Errorf("validating billable line: %w", err) } @@ -63,7 +64,7 @@ func (s *service) GenerateDetailedLines(in rating.StandardLineAccessor, opts ... } input := rate.PricerCalculateInput{ - StandardLineAccessor: in, + ProgressiveBilledLineAccessor: in, CurrencyCalculator: currency, FullProgressivelyBilledServicePeriod: fullProgressivelyBilledServicePeriod, StandardLineDiscounts: in.GetStandardLineDiscounts(), @@ -100,6 +101,31 @@ func (s *service) GenerateDetailedLines(in rating.StandardLineAccessor, opts ... return outWithTotals, nil } +type nonProgressiveBilledLine struct { + rating.StandardLineAccessor +} + +func (l nonProgressiveBilledLine) GetProgressivelyBilledServicePeriod() (timeutil.ClosedPeriod, error) { + return l.GetServicePeriod(), nil +} + +func (l nonProgressiveBilledLine) IsProgressivelyBilled() bool { + return false +} + +func (l nonProgressiveBilledLine) GetPreviouslyBilledAmount() (alpacadecimal.Decimal, error) { + return alpacadecimal.Zero, nil +} + +func (s *service) GenerateDetailedLines(in rating.StandardLineAccessor, opts ...rating.GenerateDetailedLinesOption) (rating.GenerateDetailedLinesResult, error) { + if _, ok := in.(rating.ProgressiveBilledLineAccessor); ok { + // This should never happen, but if we are feeding the rating a progresively billed line, that's the only way to prevent overcharging. + return rating.GenerateDetailedLinesResult{}, fmt.Errorf("line is progressively billed, use GenerateProgressiveBilledDetailedLines instead") + } + + return s.GenerateProgressiveBilledDetailedLines(nonProgressiveBilledLine{StandardLineAccessor: in}, opts...) +} + // UpdateTotalsFromDetailedLines is a helper method to update the totals of a line from its detailed lines. func getTotalsFromDetailedLines(in rating.GenerateDetailedLinesResult, currency currencyx.Currency) rating.GenerateDetailedLinesResult { // Calculate the line totals diff --git a/openmeter/billing/rating/service/mutator/credits_test.go b/openmeter/billing/rating/service/mutator/credits_test.go index fe44255b23..3b95cf7c08 100644 --- a/openmeter/billing/rating/service/mutator/credits_test.go +++ b/openmeter/billing/rating/service/mutator/credits_test.go @@ -406,11 +406,13 @@ func TestCreditsMutatorSkipsNonPositiveDetailedLineTotals(t *testing.T) { require.NoError(t, err) out, err := (&mutator.Credits{}).Mutate(rate.PricerCalculateInput{ - StandardLineAccessor: &billing.StandardLine{ - StandardLineBase: billing.StandardLineBase{ - CreditsApplied: billing.CreditsApplied{ - { - Amount: alpacadecimal.NewFromFloat(25), + ProgressiveBilledLineAccessor: &billing.StandardLineWithSplitLineHierarchy{ + StandardLine: &billing.StandardLine{ + StandardLineBase: billing.StandardLineBase{ + CreditsApplied: billing.CreditsApplied{ + { + Amount: alpacadecimal.NewFromFloat(25), + }, }, }, }, @@ -449,11 +451,13 @@ func TestCreditsMutatorAllocatesOnlyRemainingPayableAmount(t *testing.T) { require.NoError(t, err) out, err := (&mutator.Credits{}).Mutate(rate.PricerCalculateInput{ - StandardLineAccessor: &billing.StandardLine{ - StandardLineBase: billing.StandardLineBase{ - CreditsApplied: billing.CreditsApplied{ - { - Amount: alpacadecimal.NewFromFloat(30), + ProgressiveBilledLineAccessor: &billing.StandardLineWithSplitLineHierarchy{ + StandardLine: &billing.StandardLine{ + StandardLineBase: billing.StandardLineBase{ + CreditsApplied: billing.CreditsApplied{ + { + Amount: alpacadecimal.NewFromFloat(30), + }, }, }, }, diff --git a/openmeter/billing/rating/service/mutator/forbidunitconfig_test.go b/openmeter/billing/rating/service/mutator/forbidunitconfig_test.go index fb33348693..80b4ad294e 100644 --- a/openmeter/billing/rating/service/mutator/forbidunitconfig_test.go +++ b/openmeter/billing/rating/service/mutator/forbidunitconfig_test.go @@ -17,10 +17,12 @@ func TestForbidUnitConfigMutator(t *testing.T) { newInput := func(unitConfig *productcatalog.UnitConfig) rate.PricerCalculateInput { return rate.PricerCalculateInput{ - StandardLineAccessor: unitConfigTestLine{ - StandardLine: &billing.StandardLine{}, - price: unitPrice, - unitConfig: unitConfig, + ProgressiveBilledLineAccessor: unitConfigTestLine{ + StandardLineWithSplitLineHierarchy: billing.StandardLineWithSplitLineHierarchy{ + StandardLine: &billing.StandardLine{}, + }, + price: unitPrice, + unitConfig: unitConfig, }, Usage: &rating.Usage{ Quantity: alpacadecimal.NewFromFloat(1400), diff --git a/openmeter/billing/rating/service/mutator/unitconfig_test.go b/openmeter/billing/rating/service/mutator/unitconfig_test.go index 1c4df1ed97..cf0274b703 100644 --- a/openmeter/billing/rating/service/mutator/unitconfig_test.go +++ b/openmeter/billing/rating/service/mutator/unitconfig_test.go @@ -27,7 +27,7 @@ const ( // use it (rather than a real RateableIntent) so a single fixture can exercise the // Pre-populated cumulative case, which the charges path never produces at rating time. type unitConfigTestLine struct { - *billing.StandardLine + billing.StandardLineWithSplitLineHierarchy price *productcatalog.Price unitConfig *productcatalog.UnitConfig @@ -48,10 +48,12 @@ func mutateUnitConfig(t *testing.T, price *productcatalog.Price, unitConfig *pro t.Helper() input := rate.PricerCalculateInput{ - StandardLineAccessor: unitConfigTestLine{ - StandardLine: &billing.StandardLine{}, - price: price, - unitConfig: unitConfig, + ProgressiveBilledLineAccessor: unitConfigTestLine{ + StandardLineWithSplitLineHierarchy: billing.StandardLineWithSplitLineHierarchy{ + StandardLine: &billing.StandardLine{}, + }, + price: price, + unitConfig: unitConfig, }, Usage: &rating.Usage{ Quantity: alpacadecimal.NewFromFloat(quantity), @@ -128,10 +130,12 @@ func TestUnitConfigMutator(t *testing.T) { }) input := rate.PricerCalculateInput{ - StandardLineAccessor: unitConfigTestLine{ - StandardLine: &billing.StandardLine{}, - price: packagePrice, - unitConfig: newUnitConfig(opDivide, 1000, roundCeiling), + ProgressiveBilledLineAccessor: unitConfigTestLine{ + StandardLineWithSplitLineHierarchy: billing.StandardLineWithSplitLineHierarchy{ + StandardLine: &billing.StandardLine{}, + }, + price: packagePrice, + unitConfig: newUnitConfig(opDivide, 1000, roundCeiling), }, Usage: &rating.Usage{ Quantity: alpacadecimal.NewFromFloat(1400), @@ -148,10 +152,12 @@ func TestUnitConfigMutator(t *testing.T) { // validator makes a zero factor unreachable through normal writes, but a // corrupt import must fail the mutation instead of panicking the billing worker. input := rate.PricerCalculateInput{ - StandardLineAccessor: unitConfigTestLine{ - StandardLine: &billing.StandardLine{}, - price: unitPrice, - unitConfig: newUnitConfig(opDivide, 0, roundCeiling), + ProgressiveBilledLineAccessor: unitConfigTestLine{ + StandardLineWithSplitLineHierarchy: billing.StandardLineWithSplitLineHierarchy{ + StandardLine: &billing.StandardLine{}, + }, + price: unitPrice, + unitConfig: newUnitConfig(opDivide, 0, roundCeiling), }, Usage: &rating.Usage{ Quantity: alpacadecimal.NewFromFloat(1400), @@ -167,10 +173,12 @@ func TestUnitConfigMutator(t *testing.T) { // Re-rating reads from the raw metered quantity each run; the mutator must not // double-convert, which it guarantees by never mutating the input usage in place. input := rate.PricerCalculateInput{ - StandardLineAccessor: unitConfigTestLine{ - StandardLine: &billing.StandardLine{}, - price: unitPrice, - unitConfig: newUnitConfig(opDivide, 1000, roundCeiling), + ProgressiveBilledLineAccessor: unitConfigTestLine{ + StandardLineWithSplitLineHierarchy: billing.StandardLineWithSplitLineHierarchy{ + StandardLine: &billing.StandardLine{}, + }, + price: unitPrice, + unitConfig: newUnitConfig(opDivide, 1000, roundCeiling), }, Usage: &rating.Usage{ Quantity: alpacadecimal.NewFromFloat(1400), diff --git a/openmeter/billing/rating/service/rate/types.go b/openmeter/billing/rating/service/rate/types.go index f8c162d2db..bc04b0f07a 100644 --- a/openmeter/billing/rating/service/rate/types.go +++ b/openmeter/billing/rating/service/rate/types.go @@ -21,7 +21,7 @@ type Pricer interface { } type PricerCalculateInput struct { - rating.StandardLineAccessor + rating.ProgressiveBilledLineAccessor CurrencyCalculator currencyx.Currency FullProgressivelyBilledServicePeriod timeutil.ClosedPeriod @@ -38,8 +38,8 @@ func (i PricerCalculateInput) Validate() error { return fmt.Errorf("full service period is required") } - if i.StandardLineAccessor == nil { - return fmt.Errorf("standard line accessor is required") + if i.ProgressiveBilledLineAccessor == nil { + return fmt.Errorf("progressively billed line accessor is required") } return nil diff --git a/openmeter/billing/rating/service/testutil/ubptest.go b/openmeter/billing/rating/service/testutil/ubptest.go index ed5961fb11..02b691635a 100644 --- a/openmeter/billing/rating/service/testutil/ubptest.go +++ b/openmeter/billing/rating/service/testutil/ubptest.go @@ -56,37 +56,23 @@ type CalculationTestCase struct { UnitConfigEnabled bool } -// unitConfigLineAccessor wraps a StandardLine to expose a unit_config, which the -// StandardLine type itself never carries. -type unitConfigLineAccessor struct { - *billing.StandardLine - - unitConfig *productcatalog.UnitConfig -} - -func (l unitConfigLineAccessor) GetUnitConfig() *productcatalog.UnitConfig { - return l.unitConfig -} - -type Service interface { - GenerateDetailedLines(in rating.StandardLineAccessor, opts ...rating.GenerateDetailedLinesOption) (rating.GenerateDetailedLinesResult, error) -} - func RunCalculationTestCase(t *testing.T, tc CalculationTestCase) { t.Helper() - line := &billing.StandardLine{ - StandardLineBase: billing.StandardLineBase{ - ManagedResource: models.NewManagedResource(models.ManagedResourceInput{ - ID: "fake-line", - Name: "feature", - }), - Currency: "USD", - RateCardDiscounts: tc.Discounts, - CreditsApplied: tc.CreditsApplied, - }, - UsageBased: &billing.UsageBasedLine{ - Price: lo.ToPtr(tc.Price), + line := &billing.StandardLineWithSplitLineHierarchy{ + StandardLine: &billing.StandardLine{ + StandardLineBase: billing.StandardLineBase{ + ManagedResource: models.NewManagedResource(models.ManagedResourceInput{ + ID: "fake-line", + Name: "feature", + }), + Currency: "USD", + RateCardDiscounts: tc.Discounts, + CreditsApplied: tc.CreditsApplied, + }, + UsageBased: &billing.UsageBasedLine{ + Price: lo.ToPtr(tc.Price), + }, }, } @@ -143,14 +129,9 @@ func RunCalculationTestCase(t *testing.T, tc CalculationTestCase) { line.UsageBased.PreLinePeriodQuantity = &tc.Usage.PreLinePeriodQty line.UsageBased.MeteredPreLinePeriodQuantity = &tc.Usage.PreLinePeriodQty - var accessor rating.StandardLineAccessor = line - if tc.UnitConfig != nil { - accessor = unitConfigLineAccessor{StandardLine: line, unitConfig: tc.UnitConfig} - } - svc := service.New(service.Config{UnitConfigEnabled: tc.UnitConfigEnabled}) - res, err := svc.GenerateDetailedLines(accessor, tc.Options...) + res, err := svc.GenerateProgressiveBilledDetailedLines(line, tc.Options...) if err != nil { if tc.ExpectErrorIs != nil { require.ErrorIs(t, err, tc.ExpectErrorIs) diff --git a/openmeter/billing/service.go b/openmeter/billing/service.go index 5636406cf3..eb6a27e1c7 100644 --- a/openmeter/billing/service.go +++ b/openmeter/billing/service.go @@ -12,7 +12,6 @@ type Service interface { ProfileService CustomerOverrideService LineEngineService - SplitLineGroupService InvoiceService GatheringInvoiceService StandardInvoiceService @@ -55,13 +54,6 @@ type LineEngineService interface { OnUnsupportedCreditNote(ctx context.Context, input OnUnsupportedCreditNoteInput) error } -type SplitLineGroupService interface { - DeleteSplitLineGroup(ctx context.Context, input DeleteSplitLineGroupInput) error - UpdateSplitLineGroup(ctx context.Context, input UpdateSplitLineGroupInput) (SplitLineGroup, error) - // GetSplitLineGroupsForSubscription returns the active split-line hierarchies required for subscription sync. - GetSplitLineGroupsForSubscription(ctx context.Context, input GetLinesForSubscriptionInput) ([]SplitLineHierarchy, error) -} - type InvoiceService interface { InvoicePendingLines(ctx context.Context, input InvoicePendingLinesInput, opts ...InvoicePendingLinesOption) ([]StandardInvoice, error) diff --git a/openmeter/billing/stdinvoiceedit.go b/openmeter/billing/stdinvoiceedit.go index c7ae62ed76..2a88cf986e 100644 --- a/openmeter/billing/stdinvoiceedit.go +++ b/openmeter/billing/stdinvoiceedit.go @@ -421,23 +421,3 @@ func applyExistingLineOverrideToGatheringLine(o ExistingLineOverride, line *Gath line.RateCardDiscounts = val.Clone().UpsertCorrelationIDs() } } - -func (l LineWithInvoiceHeader) Validate() error { - if l.Line == nil { - return errors.New("line is required") - } - - if err := l.Line.Validate(); err != nil { - return fmt.Errorf("line: %w", err) - } - - if l.Invoice == nil { - return errors.New("invoice is required") - } - - if l.Invoice.GetID() == "" { - return errors.New("invoice id is required") - } - - return nil -} diff --git a/openmeter/billing/stdinvoiceline.go b/openmeter/billing/stdinvoiceline.go index b876a6d625..ab9706049a 100644 --- a/openmeter/billing/stdinvoiceline.go +++ b/openmeter/billing/stdinvoiceline.go @@ -192,10 +192,6 @@ func (i StandardLineBase) GetName() string { return i.Name } -func (i StandardLineBase) IsProgressivelyBilled() bool { - return i.SplitLineGroupID != nil -} - type SubscriptionReference struct { SubscriptionID string `json:"subscriptionID"` PhaseID string `json:"phaseID"` @@ -308,8 +304,7 @@ type StandardLine struct { UsageBased *UsageBasedLine `json:"usageBased,omitempty"` - DetailedLines DetailedLines `json:"detailedLines,omitempty"` - SplitLineHierarchy *SplitLineHierarchy `json:"progressiveLineHierarchy,omitempty"` + DetailedLines DetailedLines `json:"detailedLines,omitempty"` Discounts StandardLineDiscounts `json:"discounts,omitempty"` @@ -408,43 +403,10 @@ func (i StandardLine) GetMeteredPreLinePeriodQuantity() (*alpacadecimal.Decimal, return i.UsageBased.MeteredPreLinePeriodQuantity, nil } -func (i StandardLine) GetProgressivelyBilledServicePeriod() (timeutil.ClosedPeriod, error) { - if i.SplitLineGroupID == nil { - return timeutil.ClosedPeriod{ - From: i.Period.From, - To: i.Period.To, - }, nil - } - - if i.SplitLineHierarchy == nil { - return timeutil.ClosedPeriod{}, errors.New("split line hierarchy is required") - } - - return i.SplitLineHierarchy.Group.ServicePeriod, nil -} - -func (i StandardLine) GetPreviouslyBilledAmount() (alpacadecimal.Decimal, error) { - if i.SplitLineGroupID == nil { - return alpacadecimal.Zero, nil - } - - if i.SplitLineHierarchy == nil { - return alpacadecimal.Zero, fmt.Errorf("line[%s] does not have a progressive line hierarchy, but is a progressive billed line", i.ID) - } - - return i.SplitLineHierarchy.SumNetAmount(SumNetAmountInput{ - PeriodEndLTE: i.Period.From, - }) -} - func (i StandardLine) GetStandardLineDiscounts() StandardLineDiscounts { return i.Discounts } -func (i *StandardLine) SetSplitLineHierarchy(hierarchy *SplitLineHierarchy) { - i.SplitLineHierarchy = hierarchy -} - // ToGatheringLineBase converts the standard line to a gathering line base. // This is temporary until the full gathering invoice functionality is split. func (i StandardLine) ToGatheringLineBase() (GatheringLineBase, error) { @@ -507,7 +469,6 @@ func (i StandardLine) CloneWithoutDependencies(edits ...StandardLineEditFunction clone.DeletedAt = nil clone.ParentLineID = nil - clone.SplitLineHierarchy = nil clone.SplitLineGroupID = nil if clone.UsageBased != nil { @@ -528,11 +489,6 @@ func (i StandardLine) WithoutDBState() *StandardLine { return &i } -func (i StandardLine) WithoutSplitLineHierarchy() *StandardLine { - i.SplitLineHierarchy = nil - return &i -} - func (i StandardLine) RemoveCircularReferences() (*StandardLine, error) { clone, err := i.Clone() if err != nil { @@ -646,15 +602,6 @@ func (i StandardLine) clone(opts cloneOptions) (*StandardLine, error) { res.Discounts = i.Discounts.Clone() } - if i.SplitLineHierarchy != nil { - cloned, err := i.SplitLineHierarchy.Clone() - if err != nil { - return nil, fmt.Errorf("cloning split line hierarchy: %w", err) - } - - res.SplitLineHierarchy = lo.ToPtr(cloned) - } - return res, nil } @@ -830,6 +777,29 @@ func (i *StandardLine) SortDetailedLines() { }) } +type StandardLineWithInvoiceHeader struct { + Line *StandardLine + Invoice StandardInvoice +} + +func (l StandardLineWithInvoiceHeader) Validate() error { + if l.Line == nil { + return fmt.Errorf("line is required") + } + + if err := l.Line.Validate(); err != nil { + return fmt.Errorf("line: %w", err) + } + + // This transport only carries the parent invoice entity as a header object, + // so we validate invoice identity here instead of requiring a fully fetched invoice aggregate. + if l.Invoice.ID == "" { + return fmt.Errorf("invoice id is required") + } + + return nil +} + // helper functions for generating new lines type NewFlatFeeLineInput struct { ID string diff --git a/openmeter/billing/worker/subscriptionsync/service/persistedstate/item.go b/openmeter/billing/worker/subscriptionsync/service/persistedstate/item.go index 10f4e5bcfc..31ab37a02e 100644 --- a/openmeter/billing/worker/subscriptionsync/service/persistedstate/item.go +++ b/openmeter/billing/worker/subscriptionsync/service/persistedstate/item.go @@ -7,6 +7,7 @@ import ( "github.com/openmeterio/openmeter/openmeter/billing/charges/flatfee" "github.com/openmeterio/openmeter/openmeter/billing/charges/meta" "github.com/openmeterio/openmeter/openmeter/billing/charges/usagebased" + "github.com/openmeterio/openmeter/openmeter/billing/invoicing/legacy/splitlinegroup" "github.com/openmeterio/openmeter/pkg/models" "github.com/openmeterio/openmeter/pkg/timeutil" ) @@ -34,7 +35,7 @@ type LineGetter interface { } type SplitLineHierarchyGetter interface { - GetSplitLineHierarchy() *billing.SplitLineHierarchy + GetSplitLineHierarchy() *splitlinegroup.SplitLineHierarchy } type UsageBasedChargeGetter interface { @@ -109,7 +110,7 @@ func ItemAsLine(in Item) (billing.GenericInvoiceLine, error) { } type persistedSplitLineHierarchy struct { - hierarchy *billing.SplitLineHierarchy + hierarchy *splitlinegroup.SplitLineHierarchy } var ( @@ -120,7 +121,7 @@ var ( // newPersistedSplitLineHierarchy constructs a persisted split line hierarchy item. // Kept private so persistedstate controls construction-time validation and Item // implementations can expose non-erroring accessors. -func newPersistedSplitLineHierarchy(hierarchy *billing.SplitLineHierarchy) (persistedSplitLineHierarchy, error) { +func newPersistedSplitLineHierarchy(hierarchy *splitlinegroup.SplitLineHierarchy) (persistedSplitLineHierarchy, error) { if hierarchy == nil { return persistedSplitLineHierarchy{}, fmt.Errorf("split line hierarchy is nil") } @@ -140,7 +141,7 @@ func (i persistedSplitLineHierarchy) ServicePeriod() timeutil.ClosedPeriod { return i.hierarchy.Group.ServicePeriod } -func (i persistedSplitLineHierarchy) GetSplitLineHierarchy() *billing.SplitLineHierarchy { +func (i persistedSplitLineHierarchy) GetSplitLineHierarchy() *splitlinegroup.SplitLineHierarchy { return i.hierarchy } @@ -169,11 +170,11 @@ func (i persistedSplitLineHierarchy) HasLastLineAnnotation(annotation string) bo return child.GetAnnotations().GetBool(annotation) } -func (i persistedSplitLineHierarchy) getLastLineForAnnotations() billing.GenericInvoiceLine { +func (i persistedSplitLineHierarchy) getLastLineForAnnotations() splitlinegroup.LineHeaderAccessor { servicePeriod := i.hierarchy.Group.ServicePeriod - for _, child := range i.hierarchy.Lines { - if child.Line.GetServicePeriod().To.Equal(servicePeriod.To) && child.Line.GetDeletedAt() == nil { - return child.Line + for _, child := range i.hierarchy.Lines() { + if child.GetServicePeriod().To.Equal(servicePeriod.To) && child.GetDeletedAt() == nil { + return child } } @@ -181,7 +182,7 @@ func (i persistedSplitLineHierarchy) getLastLineForAnnotations() billing.Generic } // ItemAsSplitLineHierarchy returns the wrapped hierarchy when the persisted item is hierarchy-backed. -func ItemAsSplitLineHierarchy(in Item) (*billing.SplitLineHierarchy, error) { +func ItemAsSplitLineHierarchy(in Item) (*splitlinegroup.SplitLineHierarchy, error) { hierarchyGetter, ok := in.(SplitLineHierarchyGetter) if !ok { return nil, fmt.Errorf("persisted item does not implement split line hierarchy getter: %s", getErrorDetails(in)) diff --git a/openmeter/billing/worker/subscriptionsync/service/persistedstate/loader.go b/openmeter/billing/worker/subscriptionsync/service/persistedstate/loader.go index 88d12855f9..cc29c1c119 100644 --- a/openmeter/billing/worker/subscriptionsync/service/persistedstate/loader.go +++ b/openmeter/billing/worker/subscriptionsync/service/persistedstate/loader.go @@ -10,6 +10,7 @@ import ( "github.com/openmeterio/openmeter/openmeter/billing" "github.com/openmeterio/openmeter/openmeter/billing/charges" "github.com/openmeterio/openmeter/openmeter/billing/charges/meta" + "github.com/openmeterio/openmeter/openmeter/billing/invoicing/legacy/splitlinegroup" "github.com/openmeterio/openmeter/openmeter/streaming" "github.com/openmeterio/openmeter/openmeter/subscription" "github.com/openmeterio/openmeter/pkg/pagination" @@ -19,23 +20,28 @@ import ( type billingService interface { GetStandardLinesForSubscription(ctx context.Context, input billing.GetLinesForSubscriptionInput) (billing.StandardLines, error) GetGatheringLinesForSubscription(ctx context.Context, input billing.GetLinesForSubscriptionInput) (billing.GatheringLines, error) - GetSplitLineGroupsForSubscription(ctx context.Context, input billing.GetLinesForSubscriptionInput) ([]billing.SplitLineHierarchy, error) ListInvoices(ctx context.Context, input billing.ListInvoicesInput) (billing.ListInvoicesResponse, error) } +type splitLineGroupService interface { + GetSplitLineGroupsForSubscription(ctx context.Context, input billing.GetLinesForSubscriptionInput) ([]splitlinegroup.SplitLineHierarchy, error) +} + type chargeService interface { ListCharges(ctx context.Context, input charges.ListChargesInput) (pagination.Result[charges.Charge], error) } type Loader struct { - billingService billingService - chargeService chargeService + billingService billingService + chargeService chargeService + splitLineGroupService splitLineGroupService } -func NewLoader(billingService billingService, chargeService chargeService) Loader { +func NewLoader(billingService billingService, chargeService chargeService, splitLineGroupService splitLineGroupService) Loader { return Loader{ - billingService: billingService, - chargeService: chargeService, + billingService: billingService, + chargeService: chargeService, + splitLineGroupService: splitLineGroupService, } } @@ -59,7 +65,7 @@ func (l Loader) LoadForSubscription(ctx context.Context, subs subscription.Subsc return State{}, fmt.Errorf("getting existing gathering lines: %w", err) } - splitLineGroups, err := l.billingService.GetSplitLineGroupsForSubscription(ctx, getLinesInput) + splitLineGroups, err := l.splitLineGroupService.GetSplitLineGroupsForSubscription(ctx, getLinesInput) if err != nil { return State{}, fmt.Errorf("getting existing split line groups: %w", err) } @@ -82,7 +88,7 @@ func (l Loader) LoadForSubscription(ctx context.Context, subs subscription.Subsc return State{}, fmt.Errorf("assembling persisted invoice line items: %w", err) } - hierarchyItems, err := lo.MapErr(splitLineGroups, func(hierarchy billing.SplitLineHierarchy, _ int) (Item, error) { + hierarchyItems, err := lo.MapErr(splitLineGroups, func(hierarchy splitlinegroup.SplitLineHierarchy, _ int) (Item, error) { normalizedHierarchy, err := normalizePersistedSplitLineHierarchy(hierarchy) if err != nil { return nil, fmt.Errorf("normalizing existing split line hierarchy: %w", err) @@ -240,9 +246,14 @@ func (l Loader) loadInvoicesForSubscriptionItems(ctx context.Context, subs subsc return Invoices{}, fmt.Errorf("getting hierarchy invoice ids: %w", err) } - for _, child := range hierarchy.Lines { - invoiceIDs[child.Invoice.GetID()] = struct{}{} + for _, child := range hierarchy.StandardLines { + invoiceIDs[child.Invoice.ID] = struct{}{} + } + + if hierarchy.GatheringLine != nil { + invoiceIDs[hierarchy.GatheringLine.Invoice.ID] = struct{}{} } + default: return Invoices{}, fmt.Errorf("unsupported persisted invoice item type: %s", item.Type()) } @@ -319,7 +330,7 @@ func normalizePersistedLine(line billing.GenericInvoiceLine) (billing.GenericInv return cloned, nil } -func normalizePersistedSplitLineHierarchy(hierarchy billing.SplitLineHierarchy) (*billing.SplitLineHierarchy, error) { +func normalizePersistedSplitLineHierarchy(hierarchy splitlinegroup.SplitLineHierarchy) (*splitlinegroup.SplitLineHierarchy, error) { cloned, err := hierarchy.Clone() if err != nil { return nil, fmt.Errorf("cloning hierarchy: %w", err) @@ -327,16 +338,17 @@ func normalizePersistedSplitLineHierarchy(hierarchy billing.SplitLineHierarchy) cloned.Group.ServicePeriod = cloned.Group.ServicePeriod.Truncate(streaming.MinimumWindowSizeDuration) - for i := range cloned.Lines { - cloned.Lines[i].Line.UpdateServicePeriod(func(period *timeutil.ClosedPeriod) { - *period = period.Truncate(streaming.MinimumWindowSizeDuration) - }) + for i, line := range cloned.StandardLines { + line.ServicePeriod = line.ServicePeriod.Truncate(streaming.MinimumWindowSizeDuration) + normalizeSubscriptionReference(line.Subscription) - if invoiceAtAccessor, ok := cloned.Lines[i].Line.(billing.InvoiceAtAccessor); ok { - invoiceAtAccessor.SetInvoiceAt(invoiceAtAccessor.GetInvoiceAt().Truncate(streaming.MinimumWindowSizeDuration)) - } + cloned.StandardLines[i] = line + } - normalizeSubscriptionReference(cloned.Lines[i].Line.GetSubscriptionReference()) + if cloned.GatheringLine != nil { + cloned.GatheringLine.ServicePeriod = cloned.GatheringLine.ServicePeriod.Truncate(streaming.MinimumWindowSizeDuration) + cloned.GatheringLine.InvoiceAt = cloned.GatheringLine.InvoiceAt.Truncate(streaming.MinimumWindowSizeDuration) + normalizeSubscriptionReference(cloned.GatheringLine.Subscription) } return &cloned, nil diff --git a/openmeter/billing/worker/subscriptionsync/service/reconciler/invoiceupdater/invoiceupdate.go b/openmeter/billing/worker/subscriptionsync/service/reconciler/invoiceupdater/invoiceupdate.go index 2f5133f440..2f967e8d83 100644 --- a/openmeter/billing/worker/subscriptionsync/service/reconciler/invoiceupdater/invoiceupdate.go +++ b/openmeter/billing/worker/subscriptionsync/service/reconciler/invoiceupdater/invoiceupdate.go @@ -10,6 +10,7 @@ import ( "github.com/samber/lo" "github.com/openmeterio/openmeter/openmeter/billing" + "github.com/openmeterio/openmeter/openmeter/billing/invoicing/legacy/splitlinegroup" billinglineengine "github.com/openmeterio/openmeter/openmeter/billing/lineengine" "github.com/openmeterio/openmeter/openmeter/customer" "github.com/openmeterio/openmeter/openmeter/streaming" @@ -27,9 +28,10 @@ type QuantitySnapshotter interface { } type Config struct { - BillingService billing.Service - QuantitySnapshotter QuantitySnapshotter - Logger *slog.Logger + BillingService billing.Service + SplitLineGroupService splitlinegroup.Service + QuantitySnapshotter QuantitySnapshotter + Logger *slog.Logger } func (c Config) Validate() error { @@ -39,6 +41,10 @@ func (c Config) Validate() error { errs = append(errs, errors.New("billing service is required")) } + if c.SplitLineGroupService == nil { + errs = append(errs, errors.New("split line group service is required")) + } + if c.QuantitySnapshotter == nil { errs = append(errs, errors.New("quantity snapshotter is required")) } @@ -51,9 +57,10 @@ func (c Config) Validate() error { } type Updater struct { - billingService billing.Service - quantitySnapshotter QuantitySnapshotter - logger *slog.Logger + billingService billing.Service + quantitySnapshotter QuantitySnapshotter + splitLineGroupService splitlinegroup.Service + logger *slog.Logger } func New(config Config) (*Updater, error) { @@ -62,9 +69,10 @@ func New(config Config) (*Updater, error) { } return &Updater{ - billingService: config.BillingService, - quantitySnapshotter: config.QuantitySnapshotter, - logger: config.Logger, + billingService: config.BillingService, + quantitySnapshotter: config.QuantitySnapshotter, + splitLineGroupService: config.SplitLineGroupService, + logger: config.Logger, }, nil } @@ -252,7 +260,7 @@ type invoicePatches struct { type splitLineGroupPatches struct { deleted []models.NamespacedID - updated []billing.SplitLineGroupUpdate + updated []splitlinegroup.SplitLineGroupUpdate } func (u *Updater) parsePatches(patches []Patch) (patchesParsed, error) { @@ -358,16 +366,6 @@ 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.quantitySnapshotter.SnapshotLineQuantity(ctx, billinglineengine.SnapshotLineQuantityInput{ - Invoice: invoice, - Line: &targetStandardLine, - }) - if err != nil { - return fmt.Errorf("recalculating line[%s]: %w", targetStandardLine.ID, err) - } - - targetStandardLine = *updatedQtyLine - if ok := invoice.Lines.ReplaceByID(targetStandardLine.ID, &targetStandardLine); !ok { return fmt.Errorf("line[%s/%s] not found in the invoice, cannot update", targetStandardLine.ID, lo.FromPtrOr(targetStandardLine.ChildUniqueReferenceID, "nil")) } @@ -581,13 +579,13 @@ func (u *Updater) upsertSplitLineGroups(ctx context.Context, customerID customer } for _, groupID := range changes.deleted { - if err := u.billingService.DeleteSplitLineGroup(ctx, groupID); err != nil { + if err := u.splitLineGroupService.DeleteSplitLineGroup(ctx, groupID); err != nil { return fmt.Errorf("deleting split line group: %w", err) } } for _, group := range changes.updated { - if _, err := u.billingService.UpdateSplitLineGroup(ctx, group); err != nil { + if _, err := u.splitLineGroupService.UpdateSplitLineGroup(ctx, group); err != nil { return fmt.Errorf("upserting split line group: %w", err) } } diff --git a/openmeter/billing/worker/subscriptionsync/service/reconciler/invoiceupdater/patch.go b/openmeter/billing/worker/subscriptionsync/service/reconciler/invoiceupdater/patch.go index 177482d64b..6fb09cee31 100644 --- a/openmeter/billing/worker/subscriptionsync/service/reconciler/invoiceupdater/patch.go +++ b/openmeter/billing/worker/subscriptionsync/service/reconciler/invoiceupdater/patch.go @@ -5,6 +5,7 @@ import ( "log/slog" "github.com/openmeterio/openmeter/openmeter/billing" + "github.com/openmeterio/openmeter/openmeter/billing/invoicing/legacy/splitlinegroup" "github.com/openmeterio/openmeter/pkg/models" ) @@ -36,7 +37,7 @@ type PatchSplitLineGroupDelete struct { } type PatchSplitLineGroupUpdate struct { - TargetState billing.SplitLineGroupUpdate + TargetState splitlinegroup.SplitLineGroupUpdate } type Patch struct { @@ -122,7 +123,7 @@ func NewDeleteSplitLineGroupPatch(groupID models.NamespacedID) Patch { } } -func NewUpdateSplitLineGroupPatch(group billing.SplitLineGroupUpdate) Patch { +func NewUpdateSplitLineGroupPatch(group splitlinegroup.SplitLineGroupUpdate) Patch { return Patch{ op: PatchOpSplitLineGroupUpdate, updateSplitLineGroupPatch: PatchSplitLineGroupUpdate{ diff --git a/openmeter/billing/worker/subscriptionsync/service/reconciler/patchhelpers.go b/openmeter/billing/worker/subscriptionsync/service/reconciler/patchhelpers.go index 6fa6496bcd..45e88cf7f6 100644 --- a/openmeter/billing/worker/subscriptionsync/service/reconciler/patchhelpers.go +++ b/openmeter/billing/worker/subscriptionsync/service/reconciler/patchhelpers.go @@ -6,6 +6,7 @@ import ( "github.com/samber/lo" "github.com/openmeterio/openmeter/openmeter/billing" + "github.com/openmeterio/openmeter/openmeter/billing/invoicing/legacy/splitlinegroup" "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/streaming" @@ -28,13 +29,13 @@ func shouldSkipLinePatch(existingLine billing.GenericInvoiceLine, expectedLine b return false } -func shouldSkipHierarchyPatch(existingHierarchy *billing.SplitLineHierarchy, expectedLine billing.GatheringLine) bool { +func shouldSkipHierarchyPatch(existingHierarchy *splitlinegroup.SplitLineHierarchy, expectedLine billing.GatheringLine) bool { if expectedLine.Annotations.GetBool(billing.AnnotationSubscriptionSyncIgnore) { return true } - for _, line := range existingHierarchy.Lines { - if line.Line.GetAnnotations().GetBool(billing.AnnotationSubscriptionSyncIgnore) { + for _, line := range existingHierarchy.Lines() { + if line.GetAnnotations().GetBool(billing.AnnotationSubscriptionSyncIgnore) { return true } } diff --git a/openmeter/billing/worker/subscriptionsync/service/reconciler/patchinvoicelinehierarchy.go b/openmeter/billing/worker/subscriptionsync/service/reconciler/patchinvoicelinehierarchy.go index 1e0087b8e7..b47249327c 100644 --- a/openmeter/billing/worker/subscriptionsync/service/reconciler/patchinvoicelinehierarchy.go +++ b/openmeter/billing/worker/subscriptionsync/service/reconciler/patchinvoicelinehierarchy.go @@ -38,10 +38,13 @@ func (c *lineHierarchyPatchCollection) AddDelete(uniqueID string, existing persi if err != nil { return err } - patches := make([]invoiceupdater.Patch, 0, 1+len(group.Lines)) - for _, line := range group.Lines { - if line.Line.GetAnnotations().GetBool(billing.AnnotationSubscriptionSyncIgnore) { + lines := group.Lines() + + patches := make([]invoiceupdater.Patch, 0, 1+len(lines)) + + for _, line := range lines { + if line.GetAnnotations().GetBool(billing.AnnotationSubscriptionSyncIgnore) { return nil } } @@ -50,12 +53,12 @@ func (c *lineHierarchyPatchCollection) AddDelete(uniqueID string, existing persi patches = append(patches, invoiceupdater.NewDeleteSplitLineGroupPatch(group.Group.NamespacedID)) } - for _, line := range group.Lines { - if line.Line.GetDeletedAt() != nil { + for _, line := range lines { + if line.GetDeletedAt() != nil { continue } - patches = append(patches, invoiceupdater.NewDeleteLinePatch(line.Line.GetLineID(), line.Invoice.GetID())) + patches = append(patches, invoiceupdater.NewDeleteLinePatch(line.GetID(), line.GetInvoiceID().ID)) } if len(patches) == 0 { @@ -70,6 +73,7 @@ func (c *lineHierarchyPatchCollection) AddShrink(uniqueID string, existing persi if err != nil { return err } + expectedLine, err := target.GetExpectedLineOrErr() if err != nil { return err @@ -83,9 +87,11 @@ func (c *lineHierarchyPatchCollection) AddShrink(uniqueID string, existing persi return fmt.Errorf("shrink patch requires target end before existing hierarchy end: existing=%s..%s target=%s..%s", existingHierarchy.Group.ServicePeriod.From, existingHierarchy.Group.ServicePeriod.To, expectedLine.ServicePeriod.From, expectedLine.ServicePeriod.To) } - patches := make([]invoiceupdater.Patch, 0, len(existingHierarchy.Lines)+1) + lines := existingHierarchy.Lines() + + patches := make([]invoiceupdater.Patch, 0, len(lines)+1) - for _, child := range existingHierarchy.Lines { + for _, child := range lines { if child.Line.GetServicePeriod().To.Before(expectedLine.ServicePeriod.To) { continue }