feat(billing): CreateInvoicePendingLines charges support (#4508)

Author: mark-vass-konghq

openmeter/billing/charges/service.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/openmeterio/openmeter/openmeter/billing"
 	"github.com/openmeterio/openmeter/openmeter/billing/charges/creditpurchase"
 	"github.com/openmeterio/openmeter/openmeter/billing/charges/meta"
 	"github.com/openmeterio/openmeter/openmeter/billing/charges/models/payment"
@@ -26,6 +27,7 @@ type ChargeService interface {
 	GetByID(ctx context.Context, input GetByIDInput) (Charge, error)
 	GetByIDs(ctx context.Context, input GetByIDsInput) (Charges, error)
 	Create(ctx context.Context, input CreateInput) (Charges, error)
+	CreatePendingInvoiceLines(ctx context.Context, input CreatePendingInvoiceLinesInput) (*CreatePendingInvoiceLinesResult, error)
 	UpdateSubscriptionItemID(ctx context.Context, charge Charge, newSubscriptionItemID string) (Charge, error)
 
 	AdvanceCharges(ctx context.Context, input AdvanceChargesInput) (Charges, error)
@@ -46,6 +48,11 @@ type CreateInput struct {
 	Intents   ChargeIntents
 }
 
+type (
+	CreatePendingInvoiceLinesInput  = billing.CreatePendingInvoiceLinesInput
+	CreatePendingInvoiceLinesResult = billing.CreatePendingInvoiceLinesResult
+)
+
 func (i CreateInput) Validate() error {
 	var errs []error
 

openmeter/billing/charges/service/create.go

@@ -28,6 +28,7 @@ import (
 type chargesWithInvoiceNowActions struct {
 	charges                          charges.Charges
 	collectionAlignmentBypassedLines []invoicePendingLinesInput
+	pendingLineResults               []*billing.CreatePendingInvoiceLinesResult
 }
 
 // applyDefaultTaxCodes fills in nil TaxCodeID on each intent's TaxConfig using the namespace's
@@ -67,6 +68,27 @@ func (s *service) applyDefaultTaxCodes(ctx context.Context, namespace string, in
 }
 
 func (s *service) Create(ctx context.Context, input charges.CreateInput) (charges.Charges, error) {
+	result, err := s.create(ctx, input)
+	if err != nil {
+		return nil, err
+	}
+
+	if result == nil {
+		return nil, fmt.Errorf("result is nil")
+	}
+
+	// TODO: once we have proper state machine for credit purchases, we can remove this and make the
+	// autoAdvanceCreatedCharges handle the invoice now actions.
+	if len(result.collectionAlignmentBypassedLines) > 0 {
+		if err := s.invokeInvoiceNowOnCreate(ctx, result.collectionAlignmentBypassedLines); err != nil {
+			return nil, fmt.Errorf("invoking invoice now on create: %w", err)
+		}
+	}
+
+	return s.autoAdvanceCreatedCharges(ctx, result.charges)
+}
+
+func (s *service) create(ctx context.Context, input charges.CreateInput) (*chargesWithInvoiceNowActions, error) {
 	if input.Namespace == "" {
 		return nil, fmt.Errorf("namespace is required")
 	}
@@ -208,7 +230,7 @@ func (s *service) Create(ctx context.Context, input charges.CreateInput) (charge
 		}
 
 		// Let's generate the gathering lines for the flat fees
-		collectionAlignmentBypassedLines, err := s.createGatheringLines(ctx, gatheringLinesToCreate)
+		gatheringLineResult, err := s.createGatheringLines(ctx, gatheringLinesToCreate)
 		if err != nil {
 			return nil, err
 		}
@@ -225,26 +247,15 @@ func (s *service) Create(ctx context.Context, input charges.CreateInput) (charge
 
 		return &chargesWithInvoiceNowActions{
 			charges:                          result,
-			collectionAlignmentBypassedLines: collectionAlignmentBypassedLines,
+			collectionAlignmentBypassedLines: gatheringLineResult.collectionAlignmentBypassedLines,
+			pendingLineResults:               gatheringLineResult.pendingLineResults,
 		}, nil
 	})
 	if err != nil {
 		return nil, err
 	}
 
-	if result == nil {
-		return nil, fmt.Errorf("result is nil")
-	}
-
-	// TODO: once we have proper state machine for credit purchases, we can remove this and mek the
-	// autoAdvanceCreatedCharges handle the invoice now actions.
-	if len(result.collectionAlignmentBypassedLines) > 0 {
-		if err := s.invokeInvoiceNowOnCreate(ctx, result.collectionAlignmentBypassedLines); err != nil {
-			return nil, fmt.Errorf("invoking invoice now on create: %w", err)
-		}
-	}
-
-	return s.autoAdvanceCreatedCharges(ctx, result.charges)
+	return result, nil
 }
 
 // autoAdvanceCreatedCharges post-processes newly created charges
@@ -379,9 +390,14 @@ type invoicePendingLinesInput struct {
 	LineID     string
 }
 
-func (s *service) createGatheringLines(ctx context.Context, gatheringLinesToCreate []gatheringLineWithCustomerID) ([]invoicePendingLinesInput, error) {
+type createGatheringLinesResult struct {
+	collectionAlignmentBypassedLines []invoicePendingLinesInput
+	pendingLineResults               []*billing.CreatePendingInvoiceLinesResult
+}
+
+func (s *service) createGatheringLines(ctx context.Context, gatheringLinesToCreate []gatheringLineWithCustomerID) (createGatheringLinesResult, error) {
 	if len(gatheringLinesToCreate) == 0 {
-		return nil, nil
+		return createGatheringLinesResult{}, nil
 	}
 
 	gatheringLinesByCurrencyAndCustomer := lo.GroupBy(gatheringLinesToCreate, func(item gatheringLineWithCustomerID) currencyAndCustomerID {
@@ -391,7 +407,10 @@ func (s *service) createGatheringLines(ctx context.Context, gatheringLinesToCrea
 		}
 	})
 
-	invoiceNowLines := make([]invoicePendingLinesInput, 0, len(gatheringLinesToCreate))
+	out := createGatheringLinesResult{
+		collectionAlignmentBypassedLines: make([]invoicePendingLinesInput, 0, len(gatheringLinesToCreate)),
+		pendingLineResults:               make([]*billing.CreatePendingInvoiceLinesResult, 0, len(gatheringLinesByCurrencyAndCustomer)),
+	}
 
 	for custAndCurrency, lines := range gatheringLinesByCurrencyAndCustomer {
 		// Let's create the gathering invoice on invoicing side
@@ -403,18 +422,43 @@ func (s *service) createGatheringLines(ctx context.Context, gatheringLinesToCrea
 			}),
 		})
 		if err != nil {
-			return nil, fmt.Errorf("creating pending invoice lines for charges: %w", err)
+			return createGatheringLinesResult{}, fmt.Errorf("creating pending invoice lines for charges: %w", err)
+		}
+		if result == nil {
+			return createGatheringLinesResult{}, fmt.Errorf("creating pending invoice lines for charges: result is nil")
 		}
 
-		for idx, line := range result.Lines {
-			if lines[idx].BypassCollectionAlignment {
-				invoiceNowLines = append(invoiceNowLines, invoicePendingLinesInput{
+		out.pendingLineResults = append(out.pendingLineResults, result)
+
+		// Correlate the returned lines back to their inputs by charge ID rather than by
+		// position: billing may drop lines (e.g. zero-amount lines), which would make
+		// index-based correlation silently read BypassCollectionAlignment from the wrong line.
+		bypassChargeIDs := make(map[string]struct{}, len(lines))
+		for _, line := range lines {
+			if !line.BypassCollectionAlignment {
+				continue
+			}
+
+			if line.gatheringLine.ChargeID == nil {
+				return createGatheringLinesResult{}, fmt.Errorf("creating pending invoice lines for charges: bypass collection alignment requested for line without charge ID")
+			}
+
+			bypassChargeIDs[*line.gatheringLine.ChargeID] = struct{}{}
+		}
+
+		for _, line := range result.Lines {
+			if line.ChargeID == nil {
+				continue
+			}
+
+			if _, ok := bypassChargeIDs[*line.ChargeID]; ok {
+				out.collectionAlignmentBypassedLines = append(out.collectionAlignmentBypassedLines, invoicePendingLinesInput{
 					CustomerID: custAndCurrency.customerID,
 					LineID:     line.ID,
 				})
 			}
 		}
 	}
 
-	return invoiceNowLines, nil
+	return out, nil
 }