fix(ledger): collect credits at booking time

openmeter/billing/worker/subscriptionsync/service/sync_credittheninvoice_test.go

@@ -5079,6 +5079,8 @@ func (s *CreditThenInvoiceTestSuite) TestStandardInvoiceManualEditSync() {
 		Total:        3,
 	})
 
+	s.Require().Len(chargeBeforeEdit.Realizations.CurrentRun.CreditRealizations, 1)
+
 	updatedIntent := expectedFlatFeeIntent{
 		ServicePeriod: timeutil.ClosedPeriod{
 			From: originalLine.Period.From.Add(time.Hour),
@@ -5165,6 +5167,10 @@ func (s *CreditThenInvoiceTestSuite) TestStandardInvoiceManualEditSync() {
 	s.Equal(editedInvoice.ID, *chargeAfterEdit.Realizations.CurrentRun.InvoiceID)
 	s.Equal(updatedIntent.ServicePeriod, chargeAfterEdit.Realizations.CurrentRun.ServicePeriod)
 	s.Equal(updatedIntent.Amount, chargeAfterEdit.Realizations.CurrentRun.AmountAfterProration.InexactFloat64())
+	// The edit moves booking before the original collection. Those two credits
+	// are already allocated, so the rerate must not collect them a second time.
+	s.Equal(chargeBeforeEdit.Realizations.CurrentRun.CreditRealizations, chargeAfterEdit.Realizations.CurrentRun.CreditRealizations)
+
 	s.assertTotals(chargeAfterEdit.Realizations.CurrentRun.Totals, expectedTotalsInput{
 		Amount:       updatedIntent.Amount,
 		CreditsTotal: 2,

openmeter/ledger/chargeadapter/flatfee.go

@@ -65,19 +65,18 @@ func (h *flatFeeHandler) OnAllocateCredits(ctx context.Context, input flatfee.On
 	}
 
 	realizations, err := h.collector.CollectToAccrued(ctx, collector.CollectToAccruedInput{
-		Namespace:         input.Charge.Namespace,
-		ChargeID:          input.Charge.ID,
-		CustomerID:        intent.GetCustomerID(),
-		Annotations:       chargeAnnotationsForFlatFeeCharge(input.Charge),
-		BookedAt:          input.BookedAt,
-		SourceBalanceAsOf: intent.GetEffectiveInvoiceAt(),
-		Currency:          intent.GetCurrency().Reference(),
-		TaxCode:           lo.ToPtr(taxConfig.TaxCodeID),
-		TaxBehavior:       (*ledger.TaxBehavior)(taxConfig.Behavior),
-		SettlementMode:    intent.GetSettlementMode(),
-		ServicePeriod:     input.ServicePeriod,
-		Filters:           intent.GetCreditFilters(),
-		Amount:            input.PreTaxAmountToAllocate,
+		Namespace:      input.Charge.Namespace,
+		ChargeID:       input.Charge.ID,
+		CustomerID:     intent.GetCustomerID(),
+		Annotations:    chargeAnnotationsForFlatFeeCharge(input.Charge),
+		BookedAt:       input.BookedAt,
+		Currency:       intent.GetCurrency().Reference(),
+		TaxCode:        lo.ToPtr(taxConfig.TaxCodeID),
+		TaxBehavior:    (*ledger.TaxBehavior)(taxConfig.Behavior),
+		SettlementMode: intent.GetSettlementMode(),
+		ServicePeriod:  input.ServicePeriod,
+		Filters:        intent.GetCreditFilters(),
+		Amount:         input.PreTaxAmountToAllocate,
 	})
 	if err != nil {
 		return nil, err
@@ -237,16 +236,15 @@ func (h *flatFeeHandler) OnAllocateFiatOverageCredits(ctx context.Context, input
 	intent := input.Charge.Intent
 
 	return h.collector.CollectToReceivable(ctx, collector.CollectToReceivableInput{
-		Namespace:         input.Charge.Namespace,
-		ChargeID:          input.Charge.ID,
-		CustomerID:        intent.GetCustomerID(),
-		Annotations:       chargeAnnotationsForFlatFeeCharge(input.Charge),
-		BookedAt:          input.BookedAt,
-		SourceBalanceAsOf: input.BookedAt,
-		Currency:          currencies.NewCurrencyReference(currencyx.Code(fiatCurrency.GetFiatCode())),
-		Filters:           intent.GetCreditFilters(),
-		ServicePeriod:     input.Run.ServicePeriod,
-		Amount:            input.AmountToAllocate,
+		Namespace:     input.Charge.Namespace,
+		ChargeID:      input.Charge.ID,
+		CustomerID:    intent.GetCustomerID(),
+		Annotations:   chargeAnnotationsForFlatFeeCharge(input.Charge),
+		BookedAt:      input.BookedAt,
+		Currency:      currencies.NewCurrencyReference(currencyx.Code(fiatCurrency.GetFiatCode())),
+		Filters:       intent.GetCreditFilters(),
+		ServicePeriod: input.Run.ServicePeriod,
+		Amount:        input.AmountToAllocate,
 	})
 }
 

openmeter/ledger/chargeadapter/flatfee_test.go

@@ -780,8 +780,8 @@ func (e *flatFeeHandlerTestEnv) newAllocateCreditsInputForCharge(charge chargefl
 func (e *flatFeeHandlerTestEnv) newAssignmentInputWithMode(amount alpacadecimal.Decimal, mode productcatalog.SettlementMode) chargeflatfee.OnAllocateCreditsInput {
 	now := e.Now()
 	servicePeriod := timeutil.ClosedPeriod{
-		From: now.Add(-time.Hour),
-		To:   now,
+		From: now,
+		To:   now.Add(time.Hour),
 	}
 
 	return chargeflatfee.OnAllocateCreditsInput{
@@ -813,7 +813,7 @@ func (e *flatFeeHandlerTestEnv) newAssignmentInputWithMode(amount alpacadecimal.
 							FullServicePeriod: servicePeriod,
 							BillingPeriod:     servicePeriod,
 						},
-						InvoiceAt:             now,
+						InvoiceAt:             now.Add(-time.Hour),
 						PaymentTerm:           productcatalog.InAdvancePaymentTerm,
 						ProRating:             productcatalog.ProRatingConfig{},
 						AmountBeforeProration: amount,
@@ -932,8 +932,8 @@ func (e *flatFeeHandlerTestEnv) newAccrualInput(total alpacadecimal.Decimal) cha
 func (e *flatFeeHandlerTestEnv) newCreditsOnlyCharge(amount alpacadecimal.Decimal) chargeflatfee.Charge {
 	now := e.Now()
 	servicePeriod := timeutil.ClosedPeriod{
-		From: now.Add(-time.Hour),
-		To:   now,
+		From: now,
+		To:   now.Add(time.Hour),
 	}
 
 	charge := e.newBaseCharge(servicePeriod, amount)

openmeter/ledger/chargeadapter/origin_test.go

@@ -124,13 +124,12 @@ func (e *originTestEnv) collect(t *testing.T, spend string, amount int64) credit
 	t.Helper()
 
 	allocations, err := e.collector.CollectToAccrued(t.Context(), collector.CollectToAccruedInput{
-		Namespace:         e.Namespace,
-		CustomerID:        e.CustomerID.ID,
-		ChargeID:          spend,
-		BookedAt:          e.Now(),
-		SourceBalanceAsOf: e.Now(),
-		Currency:          e.currency.Reference(),
-		SettlementMode:    productcatalog.CreditOnlySettlementMode,
+		Namespace:      e.Namespace,
+		CustomerID:     e.CustomerID.ID,
+		ChargeID:       spend,
+		BookedAt:       e.Now(),
+		Currency:       e.currency.Reference(),
+		SettlementMode: productcatalog.CreditOnlySettlementMode,
 		ServicePeriod: timeutil.ClosedPeriod{
 			From: e.Now().Add(-time.Hour),
 			To:   e.Now(),

openmeter/ledger/chargeadapter/usagebased.go

@@ -249,16 +249,15 @@ func (h *usageBasedHandler) OnAllocateFiatOverageCredits(ctx context.Context, in
 	intent := input.Charge.Intent
 
 	return h.collector.CollectToReceivable(ctx, collector.CollectToReceivableInput{
-		Namespace:         input.Charge.Namespace,
-		ChargeID:          input.Charge.ID,
-		CustomerID:        intent.GetCustomerID(),
-		Annotations:       chargeAnnotationsForUsageBasedCharge(input.Charge),
-		BookedAt:          input.BookedAt,
-		SourceBalanceAsOf: input.BookedAt,
-		Currency:          currencies.NewCurrencyReference(currencyx.Code(fiatCurrency.GetFiatCode())),
-		Filters:           intent.GetCreditFilters(),
-		ServicePeriod:     intent.GetEffectiveServicePeriod(),
-		Amount:            input.AmountToAllocate,
+		Namespace:     input.Charge.Namespace,
+		ChargeID:      input.Charge.ID,
+		CustomerID:    intent.GetCustomerID(),
+		Annotations:   chargeAnnotationsForUsageBasedCharge(input.Charge),
+		BookedAt:      input.BookedAt,
+		Currency:      currencies.NewCurrencyReference(currencyx.Code(fiatCurrency.GetFiatCode())),
+		Filters:       intent.GetCreditFilters(),
+		ServicePeriod: intent.GetEffectiveServicePeriod(),
+		Amount:        input.AmountToAllocate,
 	})
 }
 
@@ -369,19 +368,18 @@ func (h *usageBasedHandler) OnCreditsOnlyUsageAccrued(ctx context.Context, input
 	}
 
 	realizations, err := h.collector.CollectToAccrued(ctx, collector.CollectToAccruedInput{
-		Namespace:         input.Charge.Namespace,
-		ChargeID:          input.Charge.ID,
-		CustomerID:        intent.GetCustomerID(),
-		Annotations:       chargeAnnotationsForUsageBasedCharge(input.Charge),
-		BookedAt:          input.BookedAt,
-		SourceBalanceAsOf: input.BookedAt,
-		Currency:          intent.GetCurrency().Reference(),
-		Filters:           intent.GetCreditFilters(),
-		TaxCode:           lo.ToPtr(taxConfig.TaxCodeID),
-		TaxBehavior:       (*ledger.TaxBehavior)(taxConfig.Behavior),
-		SettlementMode:    intent.GetSettlementMode(),
-		ServicePeriod:     intent.GetEffectiveServicePeriod(),
-		Amount:            input.AmountToAllocate,
+		Namespace:      input.Charge.Namespace,
+		ChargeID:       input.Charge.ID,
+		CustomerID:     intent.GetCustomerID(),
+		Annotations:    chargeAnnotationsForUsageBasedCharge(input.Charge),
+		BookedAt:       input.BookedAt,
+		Currency:       intent.GetCurrency().Reference(),
+		Filters:        intent.GetCreditFilters(),
+		TaxCode:        lo.ToPtr(taxConfig.TaxCodeID),
+		TaxBehavior:    (*ledger.TaxBehavior)(taxConfig.Behavior),
+		SettlementMode: intent.GetSettlementMode(),
+		ServicePeriod:  intent.GetEffectiveServicePeriod(),
+		Amount:         input.AmountToAllocate,
 	})
 	if err != nil {
 		return nil, err

openmeter/ledger/collector/README.md

@@ -2,7 +2,7 @@
 
 This package turns selected customer FBO credit into accrued value and, for
 custom-currency `credit_then_invoice` overage, fiat receivable coverage.
-The [FBO plan](../fbo/README.md) owns source selection, consumption, expiry
+The [FBO service](../fbo/README.md) owns source selection, consumption, expiry
 release, and credit-only advance for an uncovered amount. The collector commits
 the complete group and maps each consumption's transactions to billing
 allocations. Source order is preserved so correction and breakage can undo the
@@ -11,25 +11,15 @@ same collected amounts.
 ## Vocab
 
 - `BookedAt`: timestamp used for the ledger transactions being written.
-- `SourceBalanceAsOf`: timestamp used to decide which FBO sources are available.
 - `source`: one eligible FBO slice selected by the FBO service.
 - `source entry`: the concrete negative FBO ledger entry created by collection.
 - `allocation`: billing's collapsed record of collected credit.
 - `advance`: value moved through FBO/accrued before real credit exists to cover it.
 
-`BookedAt` and `SourceBalanceAsOf` are intentionally separate.
-
-Example:
-
-```text
-charge allocates at T1
-source balance is checked at T5
-
-BookedAt = T1
-SourceBalanceAsOf = T5
-```
-
-The transaction is booked at `T1`, but source selection can see credit and expiry state visible as of `T5`.
+Credit eligibility and consumption use the same `BookedAt`. Invoice scheduling
+and the time collection runs do not change which credits can fund that posting.
+Spendable balances also preserve credit already allocated to later postings, so
+backdating collection cannot spend it again.
 
 ## Collection Order
 

openmeter/ledger/collector/collect.go

@@ -48,18 +48,17 @@ func (c *accrualCollector) collectToAccrued(ctx context.Context, input CollectTo
 		}
 
 		inputs, err := c.fbo.Consume(ctx, fbo.ConsumeInput{
-			ChargeID:          input.ChargeID,
-			CustomerID:        customer.CustomerID{Namespace: input.Namespace, ID: input.CustomerID},
-			PostingInput:      breakage.PostingInput{TransactionGroupID: groupID, Annotations: annotations},
-			BookedAt:          input.BookedAt,
-			SourceBalanceAsOf: input.SourceBalanceAsOf,
-			Currency:          input.Currency,
-			Filters:           input.Filters,
-			Amount:            input.Amount,
-			Destination:       fbo.ConsumeDestinationAccrued,
-			AdvancePolicy:     advancePolicy,
-			TaxCode:           input.TaxCode,
-			TaxBehavior:       input.TaxBehavior,
+			ChargeID:      input.ChargeID,
+			CustomerID:    customer.CustomerID{Namespace: input.Namespace, ID: input.CustomerID},
+			PostingInput:  breakage.PostingInput{TransactionGroupID: groupID, Annotations: annotations},
+			BookedAt:      input.BookedAt,
+			Currency:      input.Currency,
+			Filters:       input.Filters,
+			Amount:        input.Amount,
+			Destination:   fbo.ConsumeDestinationAccrued,
+			AdvancePolicy: advancePolicy,
+			TaxCode:       input.TaxCode,
+			TaxBehavior:   input.TaxBehavior,
 		})
 		if err != nil {
 			return nil, err
@@ -93,16 +92,15 @@ func (c *accrualCollector) collectToReceivable(ctx context.Context, input Collec
 		groupID := ulid.Make().String()
 
 		inputs, err := c.fbo.Consume(ctx, fbo.ConsumeInput{
-			ChargeID:          input.ChargeID,
-			CustomerID:        customer.CustomerID{Namespace: input.Namespace, ID: input.CustomerID},
-			PostingInput:      breakage.PostingInput{TransactionGroupID: groupID, Annotations: annotations},
-			BookedAt:          input.BookedAt,
-			SourceBalanceAsOf: input.SourceBalanceAsOf,
-			Currency:          input.Currency,
-			Filters:           input.Filters,
-			Amount:            input.Amount,
-			Destination:       fbo.ConsumeDestinationReceivable,
-			AdvancePolicy:     fbo.AdvancePolicyNever,
+			ChargeID:      input.ChargeID,
+			CustomerID:    customer.CustomerID{Namespace: input.Namespace, ID: input.CustomerID},
+			PostingInput:  breakage.PostingInput{TransactionGroupID: groupID, Annotations: annotations},
+			BookedAt:      input.BookedAt,
+			Currency:      input.Currency,
+			Filters:       input.Filters,
+			Amount:        input.Amount,
+			Destination:   fbo.ConsumeDestinationReceivable,
+			AdvancePolicy: fbo.AdvancePolicyNever,
 		})
 		if err != nil {
 			return nil, err

openmeter/ledger/collector/collection_fbo_test.go

@@ -239,15 +239,14 @@ func TestCollectCustomerFBOReleasesBreakageInExpiryOrder(t *testing.T) {
 		To:   env.Now(),
 	}
 	allocations, err := collector.collectToAccrued(t.Context(), CollectToAccruedInput{
-		Namespace:         env.Namespace,
-		ChargeID:          testChargeID(1),
-		CustomerID:        env.CustomerID.ID,
-		BookedAt:          env.Now(),
-		SourceBalanceAsOf: env.Now(),
-		Currency:          env.CurrencyReference(),
-		SettlementMode:    productcatalog.CreditThenInvoiceSettlementMode,
-		ServicePeriod:     servicePeriod,
-		Amount:            alpacadecimal.NewFromInt(15),
+		Namespace:      env.Namespace,
+		ChargeID:       testChargeID(1),
+		CustomerID:     env.CustomerID.ID,
+		BookedAt:       env.Now(),
+		Currency:       env.CurrencyReference(),
+		SettlementMode: productcatalog.CreditThenInvoiceSettlementMode,
+		ServicePeriod:  servicePeriod,
+		Amount:         alpacadecimal.NewFromInt(15),
 	})
 	require.NoError(t, err)
 	require.Len(t, allocations, 1)
@@ -752,13 +751,12 @@ func collectToAccruedInputForTest(
 	settlementMode productcatalog.SettlementMode,
 ) CollectToAccruedInput {
 	return CollectToAccruedInput{
-		Namespace:         env.Namespace,
-		ChargeID:          chargeID,
-		CustomerID:        env.CustomerID.ID,
-		BookedAt:          env.Now(),
-		SourceBalanceAsOf: env.Now(),
-		Currency:          env.CurrencyReference(),
-		SettlementMode:    settlementMode,
+		Namespace:      env.Namespace,
+		ChargeID:       chargeID,
+		CustomerID:     env.CustomerID.ID,
+		BookedAt:       env.Now(),
+		Currency:       env.CurrencyReference(),
+		SettlementMode: settlementMode,
 		ServicePeriod: timeutil.ClosedPeriod{
 			From: env.Now().Add(-time.Hour),
 			To:   env.Now(),