refactor: split get*subscription calls

app/common/billing.go

@@ -16,6 +16,7 @@ import (
 	"github.com/openmeterio/openmeter/openmeter/billing/charges/flatfee"
 	"github.com/openmeterio/openmeter/openmeter/billing/charges/usagebased"
 	chargesworkeradvance "github.com/openmeterio/openmeter/openmeter/billing/charges/worker/advance"
+	billinglineengine "github.com/openmeterio/openmeter/openmeter/billing/lineengine"
 	"github.com/openmeterio/openmeter/openmeter/billing/rating"
 	billingratingservice "github.com/openmeterio/openmeter/openmeter/billing/rating/service"
 	billingsequence "github.com/openmeterio/openmeter/openmeter/billing/sequence"
@@ -48,9 +49,10 @@ import (
 // BillingRegistry bundles the billing and charges services. External callers that need
 // billing or charges should depend on BillingRegistry rather than individual services.
 type BillingRegistry struct {
-	Billing  billing.Service
-	Sequence billingsequence.Service
-	Charges  *ChargesRegistry
+	Billing                 billing.Service
+	LegacyBillingLineEngine *billinglineengine.Engine
+	Sequence                billingsequence.Service
+	Charges                 *ChargesRegistry
 }
 
 func (r BillingRegistry) ChargesServiceOrNil() charges.Service {
@@ -109,18 +111,34 @@ func NewBillingSequenceService(
 	})
 }
 
+func NewLegacyBillingLineEngine(
+	billingAdapter billing.Adapter,
+	billingRatingService rating.Service,
+	featureConnector feature.FeatureConnector,
+	streamingConnector streaming.Connector,
+	billingConfig config.BillingConfiguration,
+) (*billinglineengine.Engine, error) {
+	return billinglineengine.New(billinglineengine.Config{
+		SplitLineGroupAdapter:        billingAdapter,
+		RatingService:                billingRatingService,
+		FeatureService:               featureConnector,
+		StreamingConnector:           streamingConnector,
+		MaxParallelQuantitySnapshots: billingConfig.MaxParallelQuantitySnapshots,
+	})
+}
+
 // newBillingService creates the billing service and registers validators/hooks.
 // Downstream consumers should use BillingRegistry.
 func newBillingService(
 	logger *slog.Logger,
 	appService app.Service,
 	billingAdapter billing.Adapter,
 	billingRatingService rating.Service,
+	legacyBillingLineEngine *billinglineengine.Engine,
 	sequenceService billingsequence.Service,
 	customerService customer.Service,
 	featureConnector feature.FeatureConnector,
 	meterService meter.Service,
-	streamingConnector streaming.Connector,
 	eventPublisher eventbus.Publisher,
 	billingConfig config.BillingConfiguration,
 	subscriptionServices SubscriptionServiceWithWorkflow,
@@ -130,20 +148,19 @@ func newBillingService(
 	taxCodeService taxcode.Service,
 ) (billing.Service, error) {
 	service, err := billingservice.New(billingservice.Config{
-		Adapter:                      billingAdapter,
-		SequenceService:              sequenceService,
-		RatingService:                billingRatingService,
-		AppService:                   appService,
-		CustomerService:              customerService,
-		TaxCodeService:               taxCodeService,
-		FeatureService:               featureConnector,
-		Logger:                       logger,
-		MeterService:                 meterService,
-		StreamingConnector:           streamingConnector,
-		Publisher:                    eventPublisher,
-		AdvancementStrategy:          billingConfig.AdvancementStrategy,
-		FSNamespaceLockdown:          fsConfig.NamespaceLockdown,
-		MaxParallelQuantitySnapshots: billingConfig.MaxParallelQuantitySnapshots,
+		Adapter:                 billingAdapter,
+		SequenceService:         sequenceService,
+		RatingService:           billingRatingService,
+		LegacyBillingLineEngine: legacyBillingLineEngine,
+		AppService:              appService,
+		CustomerService:         customerService,
+		TaxCodeService:          taxCodeService,
+		FeatureService:          featureConnector,
+		Logger:                  logger,
+		MeterService:            meterService,
+		Publisher:               eventPublisher,
+		AdvancementStrategy:     billingConfig.AdvancementStrategy,
+		FSNamespaceLockdown:     fsConfig.NamespaceLockdown,
 	})
 	if err != nil {
 		return nil, err
@@ -184,16 +201,27 @@ func NewBillingRegistry(
 		return BillingRegistry{}, err
 	}
 
+	legacyBillingLineEngine, err := NewLegacyBillingLineEngine(
+		billingAdapter,
+		billingRatingService,
+		featureConnector,
+		streamingConnector,
+		billingConfig,
+	)
+	if err != nil {
+		return BillingRegistry{}, err
+	}
+
 	billingService, err := newBillingService(
 		logger,
 		appService,
 		billingAdapter,
 		billingRatingService,
+		legacyBillingLineEngine,
 		sequenceService,
 		customerService,
 		featureConnector,
 		meterService,
-		streamingConnector,
 		eventPublisher,
 		billingConfig,
 		subscriptionServices,
@@ -233,9 +261,10 @@ func NewBillingRegistry(
 	}
 
 	billingRegistry := BillingRegistry{
-		Billing:  billingService,
-		Sequence: sequenceService,
-		Charges:  chargesRegistry,
+		Billing:                 billingService,
+		LegacyBillingLineEngine: legacyBillingLineEngine,
+		Sequence:                sequenceService,
+		Charges:                 chargesRegistry,
 	}
 
 	// Hook registration
@@ -330,6 +359,7 @@ func NewBillingSubscriptionSyncService(logger *slog.Logger, subsServices Subscri
 	return subscriptionsyncservice.New(subscriptionsyncservice.Config{
 		SubscriptionService:     subsServices.Service,
 		BillingService:          billingRegistry.Billing,
+		LegacyBillingLineEngine: billingRegistry.LegacyBillingLineEngine,
 		ChargesService:          billingRegistry.ChargesServiceOrNil(),
 		SubscriptionSyncAdapter: subscriptionSyncAdapter,
 		FeatureFlags: subscriptionsyncservice.FeatureFlags{

openmeter/billing/adapter.go

@@ -13,7 +13,6 @@ type Adapter interface {
 	ProfileAdapter
 	CustomerOverrideAdapter
 	InvoiceLineAdapter
-	InvoiceSplitLineGroupAdapter
 	InvoiceAdapter
 	GatheringInvoiceAdapter
 	StandardInvoiceAdapter
@@ -59,7 +58,6 @@ type CustomerSynchronizationAdapter interface {
 type InvoiceLineAdapter interface {
 	UpsertInvoiceLines(ctx context.Context, input UpsertInvoiceLinesAdapterInput) ([]*StandardLine, error)
 	ListInvoiceLines(ctx context.Context, input ListInvoiceLinesAdapterInput) ([]*StandardLine, error)
-	GetLinesForSubscription(ctx context.Context, input GetLinesForSubscriptionInput) ([]LineOrHierarchy, error)
 }
 
 type InvoiceAdapter interface {
@@ -74,13 +72,15 @@ type InvoiceAdapter interface {
 }
 
 type StandardInvoiceAdapter interface {
+	GetStandardLinesForSubscription(ctx context.Context, input GetLinesForSubscriptionInput) (StandardLines, error)
 	GetStandardInvoiceById(ctx context.Context, input GetStandardInvoiceByIdInput) (StandardInvoice, error)
 	UpdateStandardInvoice(ctx context.Context, input UpdateStandardInvoiceAdapterInput) (StandardInvoice, error)
 	ListStandardInvoicesPendingAdvancement(ctx context.Context, input ListStandardInvoicesPendingAdvancementInput) ([]StandardInvoice, error)
 	CountStandardInvoicesPendingAdvancement(ctx context.Context, input CountStandardInvoicesPendingAdvancementInput) (int64, error)
 }
 
 type GatheringInvoiceAdapter interface {
+	GetGatheringLinesForSubscription(ctx context.Context, input GetLinesForSubscriptionInput) (GatheringLines, error)
 	CreateGatheringInvoice(ctx context.Context, input CreateGatheringInvoiceAdapterInput) (GatheringInvoice, error)
 	UpdateGatheringInvoice(ctx context.Context, input UpdateGatheringInvoiceAdapterInput) error
 	DeleteGatheringInvoice(ctx context.Context, input DeleteGatheringInvoiceAdapterInput) error
@@ -90,14 +90,6 @@ type GatheringInvoiceAdapter interface {
 	HardDeleteGatheringInvoiceLines(ctx context.Context, invoiceID InvoiceID, lineIDs []string) error
 }
 
-type InvoiceSplitLineGroupAdapter interface {
-	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
 }

openmeter/billing/adapter/gatheringlines.go

@@ -2,6 +2,7 @@ package billingadapter
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -11,6 +12,7 @@ import (
 	"github.com/samber/lo"
 
 	"github.com/openmeterio/openmeter/openmeter/billing"
+	"github.com/openmeterio/openmeter/openmeter/customer"
 	"github.com/openmeterio/openmeter/openmeter/ent/db"
 	"github.com/openmeterio/openmeter/openmeter/ent/db/billinginvoice"
 	"github.com/openmeterio/openmeter/openmeter/ent/db/billinginvoiceline"
@@ -25,6 +27,86 @@ import (
 	"github.com/openmeterio/openmeter/pkg/timeutil"
 )
 
+func (a *adapter) GetGatheringLinesForSubscription(ctx context.Context, in billing.GetLinesForSubscriptionInput) (billing.GatheringLines, error) {
+	if err := in.Validate(); err != nil {
+		return nil, billing.ValidationError{
+			Err: err,
+		}
+	}
+
+	return entutils.TransactingRepo(ctx, a, func(ctx context.Context, tx *adapter) (billing.GatheringLines, error) {
+		query := tx.db.BillingInvoiceLine.Query().
+			Where(billinginvoiceline.Namespace(in.Namespace)).
+			Where(billinginvoiceline.SubscriptionID(in.SubscriptionID)).
+			Where(billinginvoiceline.ParentLineIDIsNil()). // Split-line children are loaded through their hierarchy instead of as independent subscription items.
+			Where(billinginvoiceline.HasBillingInvoiceWith(
+				billinginvoice.StatusEQ(billing.StandardInvoiceStatusGathering),
+			)).
+			Where(
+				billinginvoiceline.Or(
+					billinginvoiceline.DeletedAtIsNil(),
+					billinginvoiceline.And(
+						billinginvoiceline.DeletedAtNotNil(),
+						billinginvoiceline.ManagedByEQ(billing.ManuallyManagedLine),
+					),
+				),
+			).
+			WithBillingInvoice()
+
+		if !in.IncludeChargeManaged {
+			query = query.Where(billinginvoiceline.ChargeIDIsNil())
+		}
+
+		query = tx.expandLineItems(query)
+
+		dbLines, err := query.All(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("fetching gathering lines: %w", err)
+		}
+
+		if err := errors.Join(
+			lo.Map(dbLines, func(line *db.BillingInvoiceLine, _ int) error {
+				if line.Edges.BillingInvoice == nil {
+					return fmt.Errorf("billing invoice not found for line [id=%s]", line.ID)
+				}
+
+				return nil
+			})...,
+		); err != nil {
+			return nil, err
+		}
+
+		invoiceSchemaLevelByID, err := tx.getSchemaLevelPerInvoice(ctx, customer.CustomerID{
+			Namespace: in.Namespace,
+			ID:        in.CustomerID,
+		})
+		if err != nil {
+			return nil, fmt.Errorf("getting schema level per invoice: %w", err)
+		}
+
+		dbLinesByInvoiceID := lo.GroupBy(dbLines, func(line *db.BillingInvoiceLine) string {
+			return line.Edges.BillingInvoice.ID
+		})
+
+		gatheringLines := make(billing.GatheringLines, 0, len(dbLines))
+		for invoiceID, dbLinesForInvoice := range dbLinesByInvoiceID {
+			schemaLevel, found := invoiceSchemaLevelByID[invoiceID]
+			if !found {
+				return nil, fmt.Errorf("schema level not found for invoice [id=%s]", invoiceID)
+			}
+
+			mappedLines, err := tx.mapGatheringInvoiceLinesFromDB(schemaLevel, dbLinesForInvoice)
+			if err != nil {
+				return nil, fmt.Errorf("mapping gathering lines: %w", err)
+			}
+
+			gatheringLines = append(gatheringLines, mappedLines...)
+		}
+
+		return gatheringLines, nil
+	})
+}
+
 func (a *adapter) HardDeleteGatheringInvoiceLines(ctx context.Context, invoiceID billing.InvoiceID, lineIDs []string) error {
 	if err := invoiceID.Validate(); err != nil {
 		return fmt.Errorf("validating invoice ID: %w", err)