feat: detailed line persisted flag (#4255)

Author: turip

openmeter/billing/charges/usagebased/adapter/detailedline.go

@@ -15,6 +15,7 @@ import (
 	"github.com/openmeterio/openmeter/openmeter/billing/models/stddetailedline"
 	entdb "github.com/openmeterio/openmeter/openmeter/ent/db"
 	dbchargeusagebasedrundetailedline "github.com/openmeterio/openmeter/openmeter/ent/db/chargeusagebasedrundetailedline"
+	dbchargeusagebasedruns "github.com/openmeterio/openmeter/openmeter/ent/db/chargeusagebasedruns"
 	"github.com/openmeterio/openmeter/pkg/clock"
 	"github.com/openmeterio/openmeter/pkg/framework/entutils"
 )
@@ -42,6 +43,22 @@ func (a *adapter) FetchDetailedLines(ctx context.Context, charge usagebased.Char
 			return usagebased.Charge{}, err
 		}
 
+		dbRuns, err := tx.db.ChargeUsageBasedRuns.Query().
+			Where(
+				dbchargeusagebasedruns.NamespaceEQ(charge.Namespace),
+				dbchargeusagebasedruns.ChargeIDEQ(charge.ID),
+				dbchargeusagebasedruns.IDIn(runIDs...),
+			).
+			All(ctx)
+		if err != nil {
+			return usagebased.Charge{}, err
+		}
+
+		detailedLinesPresentByRunID := make(map[string]bool, len(dbRuns))
+		for _, dbRun := range dbRuns {
+			detailedLinesPresentByRunID[dbRun.ID] = dbRun.DetailedLinesPresent
+		}
+
 		linesByRunID := make(map[string]usagebased.DetailedLines, len(charge.Realizations))
 		for _, dbLine := range dbLines {
 			line, err := mapDetailedLineFromDB(dbLine)
@@ -55,7 +72,22 @@ func (a *adapter) FetchDetailedLines(ctx context.Context, charge usagebased.Char
 		for idx, run := range charge.Realizations {
 			lines := linesByRunID[run.ID.ID]
 			slices.SortStableFunc(lines, stddetailedline.Compare[usagebased.DetailedLine])
-			charge.Realizations[idx].DetailedLines = mo.Some(lines)
+
+			detailedLinesPresent, found := detailedLinesPresentByRunID[run.ID.ID]
+			if !found {
+				charge.Realizations[idx].DetailedLines = mo.None[usagebased.DetailedLines]()
+				continue
+			}
+
+			// Safety measure: only mark detailed lines as expanded when the persisted
+			// run records that detailed lines were written at least once. Treating
+			// unknown detailed lines as an empty set can make
+			// late-event rating overcharge.
+			if detailedLinesPresent {
+				charge.Realizations[idx].DetailedLines = mo.Some(lines)
+			} else {
+				charge.Realizations[idx].DetailedLines = mo.None[usagebased.DetailedLines]()
+			}
 		}
 
 		return charge, nil
@@ -118,6 +150,17 @@ func (a *adapter) UpsertRunDetailedLines(ctx context.Context, chargeID chargesme
 			return err
 		}
 
+		if _, err := tx.db.ChargeUsageBasedRuns.Update().
+			Where(
+				dbchargeusagebasedruns.NamespaceEQ(runID.Namespace),
+				dbchargeusagebasedruns.ChargeIDEQ(chargeID.ID),
+				dbchargeusagebasedruns.ID(runID.ID),
+			).
+			SetDetailedLinesPresent(true).
+			Save(ctx); err != nil {
+			return err
+		}
+
 		if len(createBuilders) == 0 {
 			return nil
 		}

openmeter/billing/charges/usagebased/adapter/detailedline_test.go

@@ -6,7 +6,9 @@ import (
 	"time"
 
 	"github.com/alpacahq/alpacadecimal"
+	"github.com/oklog/ulid/v2"
 	"github.com/samber/lo"
+	"github.com/samber/mo"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 
@@ -18,6 +20,7 @@ import (
 	"github.com/openmeterio/openmeter/openmeter/billing/models/totals"
 	entdb "github.com/openmeterio/openmeter/openmeter/ent/db"
 	dbchargeusagebasedrundetailedline "github.com/openmeterio/openmeter/openmeter/ent/db/chargeusagebasedrundetailedline"
+	dbchargeusagebasedruns "github.com/openmeterio/openmeter/openmeter/ent/db/chargeusagebasedruns"
 	"github.com/openmeterio/openmeter/openmeter/productcatalog"
 	"github.com/openmeterio/openmeter/openmeter/testutils"
 	"github.com/openmeterio/openmeter/pkg/currencyx"
@@ -206,6 +209,220 @@ func (s *DetailedLineAdapterSuite) TestUpsertRunDetailedLinesReplacesAndSoftDele
 	s.NotNil(deletedRow.DeletedAt)
 }
 
+func (s *DetailedLineAdapterSuite) TestFetchDetailedLinesUsesDetailedLinesPresentFlag() {
+	ctx := s.T().Context()
+	namespace := "usagebased-detailedline-adapter-fetch-flag"
+	charge, runBase, _ := s.createChargeWithRun(namespace)
+
+	fetchedWithoutMaterializedLines, err := s.adapter.GetByID(ctx, usagebased.GetByIDInput{
+		ChargeID: charge.GetChargeID(),
+		Expands: chargesmeta.Expands{
+			chargesmeta.ExpandRealizations,
+			chargesmeta.ExpandDetailedLines,
+		},
+	})
+	s.Require().NoError(err)
+	s.Require().Len(fetchedWithoutMaterializedLines.Realizations, 1)
+	s.False(fetchedWithoutMaterializedLines.Realizations[0].DetailedLines.IsPresent())
+
+	s.Require().NoError(s.adapter.UpsertRunDetailedLines(ctx, charge.GetChargeID(), runBase.ID, nil))
+
+	fetchedWithMaterializedEmptyLines, err := s.adapter.GetByID(ctx, usagebased.GetByIDInput{
+		ChargeID: charge.GetChargeID(),
+		Expands: chargesmeta.Expands{
+			chargesmeta.ExpandRealizations,
+			chargesmeta.ExpandDetailedLines,
+		},
+	})
+	s.Require().NoError(err)
+	s.Require().Len(fetchedWithMaterializedEmptyLines.Realizations, 1)
+	s.True(fetchedWithMaterializedEmptyLines.Realizations[0].DetailedLines.IsPresent())
+	s.Empty(fetchedWithMaterializedEmptyLines.Realizations[0].DetailedLines.OrEmpty())
+
+	dbRun, err := s.dbClient.ChargeUsageBasedRuns.Query().
+		Where(
+			dbchargeusagebasedruns.NamespaceEQ(namespace),
+			dbchargeusagebasedruns.ID(runBase.ID.ID),
+		).
+		Only(ctx)
+	s.Require().NoError(err)
+	s.True(dbRun.DetailedLinesPresent)
+}
+
+func (s *DetailedLineAdapterSuite) TestFetchDetailedLinesDoesNotRepairDetailedLinesPresentFlagWhenRowsExist() {
+	ctx := s.T().Context()
+	namespace := "usagebased-detailedline-adapter-fetch-does-not-repair-flag"
+	charge, runBase, servicePeriod := s.createChargeWithRun(namespace)
+
+	s.Require().NoError(s.adapter.UpsertRunDetailedLines(ctx, charge.GetChargeID(), runBase.ID, usagebased.DetailedLines{
+		s.newDetailedLine(newDetailedLineInput{
+			Charge:                 charge,
+			RunID:                  runBase.ID,
+			ServicePeriod:          servicePeriod,
+			ChildUniqueReferenceID: "existing@[2026-01-01T00:00:00Z..2026-02-01T00:00:00Z]",
+			Quantity:               1,
+		}),
+	}))
+
+	_, err := s.dbClient.ChargeUsageBasedRuns.UpdateOneID(runBase.ID.ID).
+		Where(dbchargeusagebasedruns.NamespaceEQ(namespace)).
+		SetDetailedLinesPresent(false).
+		Save(ctx)
+	s.Require().NoError(err)
+
+	fetchedCharge, err := s.adapter.GetByID(ctx, usagebased.GetByIDInput{
+		ChargeID: charge.GetChargeID(),
+		Expands: chargesmeta.Expands{
+			chargesmeta.ExpandRealizations,
+			chargesmeta.ExpandDetailedLines,
+		},
+	})
+	s.Require().NoError(err)
+	s.Require().Len(fetchedCharge.Realizations, 1)
+	s.False(fetchedCharge.Realizations[0].DetailedLines.IsPresent())
+
+	dbRun, err := s.dbClient.ChargeUsageBasedRuns.Query().
+		Where(
+			dbchargeusagebasedruns.NamespaceEQ(namespace),
+			dbchargeusagebasedruns.ID(runBase.ID.ID),
+		).
+		Only(ctx)
+	s.Require().NoError(err)
+	s.False(dbRun.DetailedLinesPresent)
+}
+
+func (s *DetailedLineAdapterSuite) TestFetchDetailedLinesUsesPersistedDetailedLinesPresentFlag() {
+	ctx := s.T().Context()
+	namespace := "usagebased-detailedline-adapter-fetch-uses-persisted-flag"
+	charge, runBase, servicePeriod := s.createChargeWithRun(namespace)
+
+	s.Require().NoError(s.adapter.UpsertRunDetailedLines(ctx, charge.GetChargeID(), runBase.ID, usagebased.DetailedLines{
+		s.newDetailedLine(newDetailedLineInput{
+			Charge:                 charge,
+			RunID:                  runBase.ID,
+			ServicePeriod:          servicePeriod,
+			ChildUniqueReferenceID: "persisted@[2026-01-01T00:00:00Z..2026-02-01T00:00:00Z]",
+			Quantity:               1,
+		}),
+	}))
+
+	_, err := s.dbClient.ChargeUsageBasedRuns.UpdateOneID(runBase.ID.ID).
+		Where(dbchargeusagebasedruns.NamespaceEQ(namespace)).
+		SetDetailedLinesPresent(false).
+		Save(ctx)
+	s.Require().NoError(err)
+
+	staleCharge := charge
+	staleCharge.Realizations = usagebased.RealizationRuns{
+		{
+			RealizationRunBase: runBase,
+		},
+	}
+	staleCharge.Realizations[0].DetailedLines = mo.Some(usagebased.DetailedLines{
+		s.newDetailedLine(newDetailedLineInput{
+			Charge:                 charge,
+			RunID:                  runBase.ID,
+			ServicePeriod:          servicePeriod,
+			ChildUniqueReferenceID: "stale@[2026-01-01T00:00:00Z..2026-02-01T00:00:00Z]",
+			Quantity:               1,
+		}),
+	})
+
+	fetchedCharge, err := s.adapter.FetchDetailedLines(ctx, staleCharge)
+	s.Require().NoError(err)
+	s.Require().Len(fetchedCharge.Realizations, 1)
+	s.False(fetchedCharge.Realizations[0].DetailedLines.IsPresent())
+}
+
+func (s *DetailedLineAdapterSuite) TestFetchDetailedLinesClearsStaleDetailedLinesWhenRunMetadataIsMissing() {
+	ctx := s.T().Context()
+	namespace := "usagebased-detailedline-adapter-fetch-missing-run-metadata"
+	charge, runBase, servicePeriod := s.createChargeWithRun(namespace)
+
+	missingRunBase := runBase
+	missingRunBase.ID.ID = ulid.Make().String()
+
+	staleCharge := charge
+	staleCharge.Realizations = usagebased.RealizationRuns{
+		{
+			RealizationRunBase: missingRunBase,
+			DetailedLines: mo.Some(usagebased.DetailedLines{
+				s.newDetailedLine(newDetailedLineInput{
+					Charge:                 charge,
+					RunID:                  missingRunBase.ID,
+					ServicePeriod:          servicePeriod,
+					ChildUniqueReferenceID: "stale@[2026-01-01T00:00:00Z..2026-02-01T00:00:00Z]",
+					Quantity:               1,
+				}),
+			}),
+		},
+	}
+
+	fetchedCharge, err := s.adapter.FetchDetailedLines(ctx, staleCharge)
+	s.Require().NoError(err)
+	s.Require().Len(fetchedCharge.Realizations, 1)
+	s.False(fetchedCharge.Realizations[0].DetailedLines.IsPresent())
+}
+
+func (s *DetailedLineAdapterSuite) createChargeWithRun(namespace string) (usagebased.Charge, usagebased.RealizationRunBase, timeutil.ClosedPeriod) {
+	s.T().Helper()
+
+	featureID := ulid.Make().String()
+	customerID := s.createCustomer(namespace)
+	s.createFeature(namespace, featureID)
+
+	servicePeriod := timeutil.ClosedPeriod{
+		From: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC),
+		To:   time.Date(2026, 2, 1, 0, 0, 0, 0, time.UTC),
+	}
+
+	createdCharges, err := s.adapter.CreateCharges(s.T().Context(), usagebased.CreateChargesInput{
+		Namespace: namespace,
+		Intents: []usagebased.CreateIntent{
+			{
+				Intent: usagebased.Intent{
+					Intent: chargesmeta.Intent{
+						Name:              "usage-charge",
+						ManagedBy:         billing.SubscriptionManagedLine,
+						UniqueReferenceID: nil,
+						CustomerID:        customerID,
+						Currency:          currencyx.Code("USD"),
+						ServicePeriod:     servicePeriod,
+						FullServicePeriod: servicePeriod,
+						BillingPeriod:     servicePeriod,
+					},
+					InvoiceAt:      servicePeriod.To,
+					SettlementMode: productcatalog.CreditOnlySettlementMode,
+					FeatureKey:     featureID,
+					Price: *productcatalog.NewPriceFrom(productcatalog.UnitPrice{
+						Amount: alpacadecimal.NewFromFloat(0.1),
+					}),
+				},
+				FeatureID: featureID,
+			},
+		},
+	})
+	s.Require().NoError(err)
+	s.Require().Len(createdCharges, 1)
+
+	charge := createdCharges[0]
+	runBase, err := s.adapter.CreateRealizationRun(s.T().Context(), charge.GetChargeID(), usagebased.CreateRealizationRunInput{
+		FeatureID:       featureID,
+		Type:            usagebased.RealizationRunTypeFinalRealization,
+		StoredAtLT:      servicePeriod.To,
+		ServicePeriodTo: servicePeriod.To,
+		MeteredQuantity: alpacadecimal.NewFromInt(10),
+		Totals: totals.Totals{
+			Amount:       alpacadecimal.NewFromInt(1),
+			ChargesTotal: alpacadecimal.NewFromInt(1),
+			Total:        alpacadecimal.NewFromInt(1),
+		},
+	})
+	s.Require().NoError(err)
+
+	return charge, runBase, servicePeriod
+}
+
 func (s *DetailedLineAdapterSuite) createCustomer(namespace string) string {
 	s.T().Helper()
 

openmeter/billing/charges/usagebased/adapter/realizationrun.go

@@ -29,6 +29,7 @@ func (a *adapter) CreateRealizationRun(ctx context.Context, chargeID meta.Charge
 			SetType(input.Type).
 			SetStoredAtLt(meta.NormalizeTimestamp(input.StoredAtLT)).
 			SetServicePeriodTo(meta.NormalizeTimestamp(input.ServicePeriodTo)).
+			SetDetailedLinesPresent(false).
 			SetNillableBillingInvoiceLineID(input.LineID).
 			SetMeteredQuantity(input.MeteredQuantity)