diff --git a/AGENTS.md b/AGENTS.md index d422f4488d..59803426a6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -105,6 +105,21 @@ See [Testing](docs/development/testing.md) for harness patterns and examples. - Preserve explanatory comments during refactors unless the change makes them false or misleading. - Keep project documentation short enough that humans will read it. +- Write comments like you're explaining the code to a teammate: plain words, + concrete reasons, no formal filler. Keep the useful explanation: + + Bad: + ```go + // scopedEntries applies the account, route, provenance, and whole-history + // exclusions shared by both branches. AsOf is left to each branch: filtering + // here would hide future postings whose balance impact must be preserved. + ``` + + Good: + ```go + // Apply the shared filters, but keep entries after AsOf: we need those to see + // how low the balance gets later. + ``` ## Further guidance diff --git a/openmeter/ledger/README.md b/openmeter/ledger/README.md index f67d14a9d7..d01d133018 100644 --- a/openmeter/ledger/README.md +++ b/openmeter/ledger/README.md @@ -24,7 +24,9 @@ within a namespace. The historical ledger stores immutable transaction groups, transactions, and entries. Balances are projections over those entries; they are not mutable -facts stored independently from the journal. +facts stored independently from the journal. Account balances sum the selected +entries. Minimum balance buckets take the lowest balance from `AsOf` onward; +[FBO spendable balances](fbo/README.md) apply the credit and breakage policy. ## Ownership and boundaries @@ -35,9 +37,9 @@ facts stored independently from the journal. - `chargeadapter` translates charge lifecycle events into ledger templates. Charges decide when an effect is due; the ledger decides how that effect is represented and validated. See the [charges domain](../billing/charges/README.md). -- The collector owns source selection and correction unwind order. Callers - provide the target amount and attribution; they must not recreate collection - policy. +- [FBO](fbo/README.md) owns consumption source selection, expiry release, and + advance coordination. The collector maps consumptions to billing allocations, + commits their group, and owns correction unwind order. - `customerbalance` is a customer-facing projection over booked ledger state, breakage, and not-yet-booked charge impacts. It does not own posting, collection, or correction rules. diff --git a/openmeter/ledger/balance.go b/openmeter/ledger/balance.go index b2c4e722f5..3959a209e0 100644 --- a/openmeter/ledger/balance.go +++ b/openmeter/ledger/balance.go @@ -71,6 +71,36 @@ type BalanceBucket struct { PendingAmount alpacadecimal.Decimal } +// MinimumBalanceBucketQuery returns each bucket's lowest balance from AsOf +// through all recorded future postings. AsOf is inclusive; simultaneous postings +// are netted before comparing balances. A future-only bucket starts at zero. +type MinimumBalanceBucketQuery struct { + BalanceBucketQuery + + // ExcludeFutureAnnotationFilters omits matching movements strictly after + // AsOf. The opening balance still includes them. + ExcludeFutureAnnotationFilters map[string]string +} + +var _ models.Validator = MinimumBalanceBucketQuery{} + +func (q MinimumBalanceBucketQuery) Validate() error { + var errs []error + if err := q.BalanceBucketQuery.Validate(); err != nil { + errs = append(errs, err) + } + + if q.Filters.AsOf == nil { + errs = append(errs, errors.New("as of is required")) + } + + if q.Filters.After != nil || q.Filters.BookedAtPeriod != nil { + errs = append(errs, errors.New("minimum balance cannot truncate the posting history")) + } + + return models.NewNillableGenericValidationError(errors.Join(errs...)) +} + // GetBalancesAtBoundariesInput describes independent persisted balance boundaries. // Results follow Queries order, including zero balances for empty scopes. type GetBalancesAtBoundariesInput struct { @@ -101,4 +131,5 @@ type BalanceQuerier interface { GetAccountBalance(ctx context.Context, account Account, route RouteFilter, query BalanceQuery) (Balance, error) GetSubAccountBalance(ctx context.Context, subAccount SubAccount, query BalanceQuery) (Balance, error) GetBalanceBuckets(ctx context.Context, query BalanceBucketQuery) ([]BalanceBucket, error) + GetMinimumBalanceBuckets(ctx context.Context, query MinimumBalanceBucketQuery) ([]BalanceBucket, error) } diff --git a/openmeter/ledger/collector/README.md b/openmeter/ledger/collector/README.md index 4f64904b61..110917dd46 100644 --- a/openmeter/ledger/collector/README.md +++ b/openmeter/ledger/collector/README.md @@ -2,15 +2,17 @@ This package turns selected customer FBO credit into accrued value and, for custom-currency `credit_then_invoice` overage, fiat receivable coverage. -Credit-only accrual asks [advance](../advance/README.md) to create an advance for -an uncovered amount. Source order is preserved so correction and breakage can -undo the same collected amounts. +The [FBO plan](../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 +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 spendable FBO slice selected by the collector. +- `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. @@ -42,7 +44,8 @@ stable cursor asc Non-expiring credit sorts after expiring credit with the same priority. -This order must match breakage release order. If the collector consumes an expiring source, it also asks breakage to release the matching planned breakage for that same source. +This order must match breakage release order. Consuming an expiring source also +releases the matching planned breakage for that source. ## Forward Collection Example diff --git a/openmeter/ledger/collector/collect.go b/openmeter/ledger/collector/collect.go index 5f208ca7dc..3811182645 100644 --- a/openmeter/ledger/collector/collect.go +++ b/openmeter/ledger/collector/collect.go @@ -2,20 +2,18 @@ package collector import ( "context" - "errors" "fmt" "maps" "github.com/alpacahq/alpacadecimal" "github.com/oklog/ulid/v2" - "github.com/samber/lo" "github.com/openmeterio/openmeter/openmeter/billing/charges/models/creditrealization" "github.com/openmeterio/openmeter/openmeter/billing/charges/models/ledgertransaction" "github.com/openmeterio/openmeter/openmeter/customer" "github.com/openmeterio/openmeter/openmeter/ledger" - "github.com/openmeterio/openmeter/openmeter/ledger/advance" "github.com/openmeterio/openmeter/openmeter/ledger/breakage" + "github.com/openmeterio/openmeter/openmeter/ledger/fbo" "github.com/openmeterio/openmeter/openmeter/ledger/transactions" "github.com/openmeterio/openmeter/openmeter/productcatalog" "github.com/openmeterio/openmeter/pkg/framework/transaction" @@ -25,104 +23,87 @@ import ( type accrualCollector struct { ledger ledger.Ledger - advance advance.Service - deps transactions.ResolverDependencies - breakage breakage.Service - accountLocker ledger.AccountLocker + fbo fbo.Service transactionManager transaction.Creator } type collectedInputs []ledger.TransactionInput func (c *accrualCollector) collectToAccrued(ctx context.Context, input CollectToAccruedInput) (creditrealization.CreateAllocationInputs, error) { - run := func(ctx context.Context) (creditrealization.CreateAllocationInputs, error) { + return transaction.Run(ctx, c.transactionManager, func(ctx context.Context) (creditrealization.CreateAllocationInputs, error) { if input.Amount.IsZero() { return nil, nil } - groupAnnotations := input.Annotations - if groupAnnotations == nil { - groupAnnotations = ledger.ChargeAnnotations(models.NamespacedID{ - Namespace: input.Namespace, - ID: input.ChargeID, - }) + annotations := input.Annotations + if annotations == nil { + annotations = ledger.ChargeAnnotations(models.NamespacedID{Namespace: input.Namespace, ID: input.ChargeID}) } - posting := breakage.PostingInput{ - TransactionGroupID: ulid.Make().String(), - Annotations: groupAnnotations, + groupID := ulid.Make().String() + + advancePolicy := fbo.AdvancePolicyNever + if input.SettlementMode == productcatalog.CreditOnlySettlementMode { + advancePolicy = fbo.AdvancePolicyOnShortfall } - inputs, err := c.resolveCollectedInputs(ctx, posting, input) + 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, + }) if err != nil { return nil, err } - // Credit-only: if the wallet didn't cover the full accrual, issue advance and - // move that slice through the advance-to-accrued path. - if shortfall := input.Amount.Sub(collectedInputs(inputs).collectedFBOAmount()); c.shouldAdvanceShortfall(input, shortfall) { - advanceInputs, err := c.advance.PlanIssue(ctx, advance.IssueInput{ - CustomerID: c.customerID(input), - ChargeID: input.ChargeID, - At: input.BookedAt, - Amount: shortfall, - Currency: input.Currency, - Filters: input.Filters, - TaxCode: input.TaxCode, - TaxBehavior: input.TaxBehavior, - }) - if err != nil { - return nil, err - } - - inputs = append(inputs, advanceInputs...) - } - if len(inputs) == 0 { return nil, nil } - for i, txInput := range inputs { - if txInput != nil { - inputs[i] = transactions.WithAnnotations(txInput, groupAnnotations) - } - } - - transactionGroup, err := c.ledger.CommitGroup(ctx, ledger.WithGroupID(transactions.GroupInputs( - input.Namespace, - groupAnnotations, - inputs..., - ), posting.TransactionGroupID)) + groupInput := ledger.WithGroupID(transactions.GroupInputs(input.Namespace, annotations, inputs...), groupID) + group, err := c.ledger.CommitGroup(ctx, groupInput) if err != nil { return nil, fmt.Errorf("commit ledger transaction group: %w", err) } - return collectedInputs(inputs).toCreditRealizations(input.ServicePeriod, transactionGroup.ID().ID), nil - } - - return transaction.Run(ctx, c.transactionManager, run) + return collectedInputs(inputs).toCreditRealizations(input.ServicePeriod, group.ID().ID), nil + }) } func (c *accrualCollector) collectToReceivable(ctx context.Context, input CollectToReceivableInput) (creditrealization.CreateAllocationInputs, error) { - run := func(ctx context.Context) (creditrealization.CreateAllocationInputs, error) { + return transaction.Run(ctx, c.transactionManager, func(ctx context.Context) (creditrealization.CreateAllocationInputs, error) { if input.Amount.IsZero() { return nil, nil } - groupAnnotations := input.Annotations - if groupAnnotations == nil { - groupAnnotations = ledger.ChargeAnnotations(models.NamespacedID{ - Namespace: input.Namespace, - ID: input.ChargeID, - }) - } - - posting := breakage.PostingInput{ - TransactionGroupID: ulid.Make().String(), - Annotations: groupAnnotations, - } - - inputs, err := c.resolveCoveredReceivableInputs(ctx, posting, input) + annotations := input.Annotations + if annotations == nil { + annotations = ledger.ChargeAnnotations(models.NamespacedID{Namespace: input.Namespace, ID: input.ChargeID}) + } + + 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, + }) if err != nil { return nil, err } @@ -131,262 +112,14 @@ func (c *accrualCollector) collectToReceivable(ctx context.Context, input Collec return nil, nil } - for i, txInput := range inputs { - if txInput != nil { - inputs[i] = transactions.WithAnnotations(txInput, groupAnnotations) - } - } - - transactionGroup, err := c.ledger.CommitGroup(ctx, ledger.WithGroupID(transactions.GroupInputs( - input.Namespace, - groupAnnotations, - inputs..., - ), posting.TransactionGroupID)) + groupInput := ledger.WithGroupID(transactions.GroupInputs(input.Namespace, annotations, inputs...), groupID) + group, err := c.ledger.CommitGroup(ctx, groupInput) if err != nil { return nil, fmt.Errorf("commit ledger transaction group: %w", err) } - return collectedInputs(inputs).toCreditRealizations(input.ServicePeriod, transactionGroup.ID().ID), nil - } - - return transaction.Run(ctx, c.transactionManager, run) -} - -func (c *accrualCollector) resolveCoveredReceivableInputs(ctx context.Context, posting breakage.PostingInput, input CollectToReceivableInput) ([]ledger.TransactionInput, error) { - if err := ledger.ValidateTransactionAmount(input.Amount); err != nil { - return nil, fmt.Errorf("amount: %w", err) - } - - selections, err := c.collectCustomerFBOSelections( - ctx, - customer.CustomerID{Namespace: input.Namespace, ID: input.CustomerID}, - input.Currency, - ledger.Route{Filters: input.Filters}, - input.Amount, - input.SourceBalanceAsOf, - ) - if err != nil { - return nil, fmt.Errorf("collect customer FBO: %w", err) - } - - if len(selections) == 0 { - return nil, nil - } - - for i := range selections { - selections[i].collectionOriginID = lo.ToPtr(ulid.Make().String()) - } - - sources := fboCollectionSelections(selections).postingAmounts(&input.ChargeID) - inputs, err := transactions.ResolveTransactions( - ctx, - c.deps, - transactions.ResolutionScope{ - CustomerID: customer.CustomerID{Namespace: input.Namespace, ID: input.CustomerID}, - Namespace: input.Namespace, - }, - transactions.CoverCustomerReceivableTemplate{ - At: input.BookedAt, - Currency: input.Currency, - Sources: sources, - }, - ) - if err != nil { - return nil, fmt.Errorf("resolve transactions: %w", err) - } - - inputs, err = c.resolveCollectionBreakageInputs(ctx, collectionBreakageInput{ - Posting: posting, - ChargeID: input.ChargeID, - Selections: selections, - Postings: inputs, - }) - if err != nil { - return nil, err - } - - return inputs, nil -} - -func (c *accrualCollector) resolveCollectedInputs(ctx context.Context, posting breakage.PostingInput, input CollectToAccruedInput) ([]ledger.TransactionInput, error) { - if err := ledger.ValidateTransactionAmount(input.Amount); err != nil { - return nil, fmt.Errorf("amount: %w", err) - } - - if err := input.Currency.Validate(); err != nil { - return nil, fmt.Errorf("currency: %w", err) - } - - if input.Currency.IsCustom() && !input.Currency.IsResolved() { - return nil, fmt.Errorf("currency: custom currency must be resolved") - } - - selections, err := c.collectCustomerFBOSelections(ctx, c.customerID(input), input.Currency, ledger.Route{Filters: input.Filters}, input.Amount, input.SourceBalanceAsOf) - if err != nil { - return nil, fmt.Errorf("collect customer FBO: %w", err) - } - - if len(selections) == 0 { - return nil, nil - } - - for i := range selections { - selections[i].collectionOriginID = lo.ToPtr(ulid.Make().String()) - } - - sources := fboCollectionSelections(selections).postingAmounts(&input.ChargeID) - inputs, err := transactions.ResolveTransactions( - ctx, - c.deps, - c.resolutionScope(input), - transactions.TransferCustomerFBOToAccruedTemplate{ - At: input.BookedAt, - Currency: input.Currency, - TaxCode: input.TaxCode, - TaxBehavior: input.TaxBehavior, - Sources: sources, - }, - ) - if err != nil { - return nil, fmt.Errorf("resolve transactions: %w", err) - } - - inputs, err = c.resolveCollectionBreakageInputs(ctx, collectionBreakageInput{ - Posting: posting, - ChargeID: input.ChargeID, - Selections: selections, - Postings: inputs, + return collectedInputs(inputs).toCreditRealizations(input.ServicePeriod, group.ID().ID), nil }) - if err != nil { - return nil, err - } - - return inputs, nil -} - -type sourcePostingIDs struct { - transactionID string - entryID string -} - -type collectionBreakageInput struct { - Posting breakage.PostingInput - ChargeID string - Selections []fboCollectionSelection - Postings []ledger.TransactionInput -} - -var _ models.Validator = collectionBreakageInput{} - -func (i collectionBreakageInput) Validate() error { - var errs []error - - if err := i.Posting.Validate(); err != nil { - errs = append(errs, err) - } - - if i.ChargeID == "" { - errs = append(errs, errors.New("charge id is required")) - } - - return models.NewNillableGenericValidationError(errors.Join(errs...)) -} - -func (c *accrualCollector) resolveCollectionBreakageInputs(ctx context.Context, input collectionBreakageInput) ([]ledger.TransactionInput, error) { - if err := input.Validate(); err != nil { - return nil, err - } - - inputs := input.Postings - - // Capture concrete source IDs while the resolved postings are still in hand. - // A collection origin identifies exactly one selected FBO debit in this group. - sourcesByOrigin := make(map[string]sourcePostingIDs, len(input.Selections)) - for idx, input := range inputs { - identified, err := ledger.PreassignIDs(input) - if err != nil { - return nil, err - } - - inputs[idx] = identified - - for _, entry := range identified.EntryInputs() { - if entry.PostingAddress().AccountType() != ledger.AccountTypeCustomerFBO || !entry.Amount().IsNegative() { - continue - } - - originID := lo.FromPtr(entry.Provenance().CollectionOriginID) - if _, exists := sourcesByOrigin[originID]; exists { - return nil, fmt.Errorf("collection origin %s has multiple FBO source entries", originID) - } - - sourcesByOrigin[originID] = sourcePostingIDs{transactionID: identified.AssignedID(), entryID: entry.AssignedID()} - } - } - - releaseRemainingByPlanID := make(map[string]alpacadecimal.Decimal) - - for _, selection := range input.Selections { - if selection.source.breakagePlan == nil { - continue - } - - plan := *selection.source.breakagePlan - remaining, ok := releaseRemainingByPlanID[plan.ID.ID] - if !ok { - remaining = plan.OpenAmount - } - - // Legacy source-less plans can reserve multiple selected source slices, - // so guard the aggregate release amount before writing release records. - if selection.amount.GreaterThan(remaining) { - return nil, fmt.Errorf("breakage release amount %s exceeds remaining plan amount %s for plan %s", selection.amount, remaining, plan.ID.ID) - } - - releaseRemainingByPlanID[plan.ID.ID] = remaining.Sub(selection.amount) - - source, ok := sourcesByOrigin[lo.FromPtr(selection.collectionOriginID)] - if !ok { - return nil, fmt.Errorf("selected credit has no resolved FBO source posting") - } - - releaseInput, err := c.breakage.ReleasePlan(ctx, breakage.ReleasePlanInput{ - PostingInput: input.Posting, - Plan: plan, - Amount: selection.amount, - SourceKind: breakage.SourceKindUsage, - SourceChargeID: selection.source.sourceChargeID, - SpendChargeID: &input.ChargeID, - CollectionOriginID: selection.collectionOriginID, - SourceTransactionID: &source.transactionID, - SourceEntryID: &source.entryID, - }) - if err != nil { - return nil, fmt.Errorf("resolve breakage release: %w", err) - } - - inputs = append(inputs, releaseInput) - } - - return inputs, nil -} - -func (c *accrualCollector) shouldAdvanceShortfall(input CollectToAccruedInput, shortfall alpacadecimal.Decimal) bool { - return input.SettlementMode == productcatalog.CreditOnlySettlementMode && shortfall.IsPositive() -} - -func (c *accrualCollector) resolutionScope(input CollectToAccruedInput) transactions.ResolutionScope { - return transactions.ResolutionScope{ - CustomerID: c.customerID(input), - Namespace: input.Namespace, - } -} - -func (c *accrualCollector) customerID(input CollectToAccruedInput) customer.CustomerID { - return customer.CustomerID{ - Namespace: input.Namespace, - ID: input.CustomerID, - } } func (i collectedInputs) toCreditRealizations(servicePeriod timeutil.ClosedPeriod, transactionGroupID string) creditrealization.CreateAllocationInputs { @@ -437,20 +170,3 @@ func (i collectedInputs) toCreditRealizations(servicePeriod timeutil.ClosedPerio return out } - -func (i collectedInputs) collectedFBOAmount() alpacadecimal.Decimal { - total := alpacadecimal.Zero - for _, input := range i { - if input == nil { - continue - } - - for _, entry := range input.EntryInputs() { - if entry.Amount().IsNegative() && entry.PostingAddress().AccountType() == ledger.AccountTypeCustomerFBO { - total = total.Add(entry.Amount().Abs()) - } - } - } - - return total -} diff --git a/openmeter/ledger/collector/collection_fbo_test.go b/openmeter/ledger/collector/collection_fbo_test.go index d58dcc1b7c..aec900e36d 100644 --- a/openmeter/ledger/collector/collection_fbo_test.go +++ b/openmeter/ledger/collector/collection_fbo_test.go @@ -18,9 +18,11 @@ import ( advancetestutils "github.com/openmeterio/openmeter/openmeter/ledger/advance/testutils" ledgerbreakage "github.com/openmeterio/openmeter/openmeter/ledger/breakage" ledgerbreakageadapter "github.com/openmeterio/openmeter/openmeter/ledger/breakage/adapter" + "github.com/openmeterio/openmeter/openmeter/ledger/fbo" ledgertestutils "github.com/openmeterio/openmeter/openmeter/ledger/testutils" "github.com/openmeterio/openmeter/openmeter/ledger/transactions" "github.com/openmeterio/openmeter/openmeter/productcatalog" + omtestutils "github.com/openmeterio/openmeter/openmeter/testutils" "github.com/openmeterio/openmeter/pkg/currencyx" "github.com/openmeterio/openmeter/pkg/framework/transaction" "github.com/openmeterio/openmeter/pkg/timeutil" @@ -56,21 +58,8 @@ func TestCollectCustomerFBOSeparatesManagedCurrenciesWithSameCode(t *testing.T) env.CustomCurrency = &beta betaFBO := fundPriority(t, env, 1, 60) - // then: a same-code breakage plan from beta is not attached to alpha sources. - env.CustomCurrency = &alpha - require.Empty(t, reserveSourcesForBreakagePlan( - []fboCollectionSource{{ - address: alphaFBO.Address(), - available: alpacadecimal.NewFromInt(60), - }}, - ledgerbreakage.Plan{ - FBOAddress: betaFBO.Address(), - }, - env.CurrencyReference(), - ledger.Route{}, - )) - // when: collection targets only the alpha managed currency. + env.CustomCurrency = &alpha _, err := collector.collectToAccrued(t.Context(), collectToAccruedInputForTest( env, "spend-alpha", @@ -107,25 +96,32 @@ func TestCollectCustomerFBOUsesSubAccountIDTieBreaker(t *testing.T) { require.Equal(t, expected[1].Address().SubAccountID(), sources[1].Address.SubAccountID()) } -func TestCollectCustomerFBOUsesAsOfBalance(t *testing.T) { +func TestCollectCustomerFBOPreservesFutureSpending(t *testing.T) { env := ledgertestutils.NewIntegrationEnv(t, "collector") collector := newTestAccrualCollector(t, env) + // given: thirty of fifty credits are already allocated to future usage. source := fundPriority(t, env, 1, 50) bookFutureFBOCollection(t, env, 1, 30, env.Now().AddDate(0, 0, 1)) + // when: source selection runs before that usage becomes effective. currentSources, err := collectCustomerFBOForTest(t, env, collector, alpacadecimal.NewFromInt(50), env.Now()) require.NoError(t, err) require.Len(t, currentSources, 1) require.Equal(t, source.Address().SubAccountID(), currentSources[0].Address.SubAccountID()) - require.True(t, alpacadecimal.NewFromInt(50).Equal(currentSources[0].Amount), "current amount: %s", currentSources[0].Amount) + // then: account balance remains historical, but collection protects the future allocation. + require.Equal(t, 20.0, currentSources[0].Amount.InexactFloat64()) + balance, err := env.Deps.HistoricalLedger.GetAccountBalance(t.Context(), env.CustomerAccounts.FBOAccount, + ledger.RouteFilter{Currency: env.CurrencyReference()}, ledger.BalanceQuery{AsOf: lo.ToPtr(env.Now())}) + require.NoError(t, err) + require.Equal(t, 50.0, balance.InexactFloat64()) futureAsOf := env.Now().AddDate(0, 0, 1) futureSources, err := collectCustomerFBOForTest(t, env, collector, alpacadecimal.NewFromInt(50), futureAsOf) require.NoError(t, err) require.Len(t, futureSources, 1) require.Equal(t, source.Address().SubAccountID(), futureSources[0].Address.SubAccountID()) - require.True(t, alpacadecimal.NewFromInt(20).Equal(futureSources[0].Amount), "future amount: %s", futureSources[0].Amount) + require.Equal(t, 20.0, futureSources[0].Amount.InexactFloat64()) } func TestCollectCustomerFBOFiltersByFeatureEligibility(t *testing.T) { @@ -169,7 +165,7 @@ func TestCollectCustomerFBOFiltersByFeatureEligibility(t *testing.T) { func TestCollectCustomerFBOFiltersBreakageByFeatureEligibility(t *testing.T) { env := ledgertestutils.NewIntegrationEnv(t, "collector") breakageService := newTestBreakageService(t, env) - collector := newTestAccrualCollectorWithBreakage(t, env, breakageService) + collector := newTestAccrualCollector(t, env) expiresAt := env.Now().Add(10 * time.Hour) bookExpiringCreditWithFeatures(t, env, breakageService, 1, 10, nil, nil, expiresAt) @@ -233,7 +229,7 @@ func TestCollectCustomerFBOUsesPriorityBeforeFeatureRestriction(t *testing.T) { func TestCollectCustomerFBOReleasesBreakageInExpiryOrder(t *testing.T) { env := ledgertestutils.NewIntegrationEnv(t, "collector") breakageService := newTestBreakageService(t, env) - collector := newTestAccrualCollectorWithBreakage(t, env, breakageService) + collector := newTestAccrualCollector(t, env) firstPlanID := bookExpiringCredit(t, env, breakageService, 1, 10, env.Now().Add(10*time.Hour)) secondPlanID := bookExpiringCredit(t, env, breakageService, 1, 15, env.Now().Add(15*time.Hour)) @@ -272,7 +268,7 @@ func TestCollectCustomerFBOReleasesBreakageInExpiryOrder(t *testing.T) { func TestCollectCustomerFBOBreakageReleaseTracksSpendOnFBOAndSourceOnBreakage(t *testing.T) { env := ledgertestutils.NewIntegrationEnv(t, "collector") breakageService := newTestBreakageService(t, env) - collector := newTestAccrualCollectorWithBreakage(t, env, breakageService) + collector := newTestAccrualCollector(t, env) // given: // - two same-route expiring credit sources with separate breakage plans @@ -336,7 +332,7 @@ func TestCollectCustomerFBOBreakageReleaseTracksSpendOnFBOAndSourceOnBreakage(t func TestCollectCustomerFBOBreakageReleaseUsesPlanSourceBeforeBucketCursor(t *testing.T) { env := ledgertestutils.NewIntegrationEnv(t, "collector") breakageService := newTestBreakageService(t, env) - collector := newTestAccrualCollectorWithBreakage(t, env, breakageService) + collector := newTestAccrualCollector(t, env) // given: // - two same-route expiring credit sources sharing one FBO sub-account @@ -504,25 +500,6 @@ func TestCollectToAccruedCustomCurrencyCreditThenInvoiceDoesNotCreateExposure(t requireReceivableBalanceBuckets(t, env, map[string]float64{}) } -func newTestAccrualCollector(t testing.TB, env *ledgertestutils.IntegrationEnv) *accrualCollector { - t.Helper() - - breakageService := ledgerbreakage.NewNoopService() - - return &accrualCollector{ - advance: advancetestutils.NewService(t, env.Deps, breakageService), - breakage: breakageService, - ledger: env.Deps.HistoricalLedger, - deps: transactions.ResolverDependencies{ - AccountService: env.Deps.ResolversService, - AccountCatalog: env.Deps.AccountService, - BalanceQuerier: env.Deps.HistoricalLedger, - }, - accountLocker: env.Deps.AccountService, - transactionManager: enttx.NewCreator(env.DB), - } -} - func collectCustomerFBOForTest( t *testing.T, env *ledgertestutils.IntegrationEnv, @@ -551,29 +528,60 @@ func collectCustomerFBOForFeatureForTest( filters.Features = []string{featureKey} } - selections, err := collector.collectCustomerFBOSelections(ctx, env.CustomerID, env.CurrencyReference(), ledger.Route{Filters: filters}, target, asOf) + sources, err := collector.fbo.ListSources(ctx, fbo.SourceQuery{ + CustomerID: env.CustomerID, + Currency: env.CurrencyReference(), + Filters: filters, + AsOf: asOf, + }) if err != nil { return nil, err } - return fboCollectionSelections(selections).postingAmounts(nil), nil + remaining := target + amounts := make([]transactions.PostingAmount, 0, len(sources)) + for _, source := range sources { + if !remaining.IsPositive() { + break + } + + amount := source.Amount + if amount.GreaterThan(remaining) { + amount = remaining + } + + amounts = append(amounts, transactions.PostingAmount{Address: source.Address, Amount: amount}) + remaining = remaining.Sub(amount) + } + + return amounts, nil }) } -func newTestAccrualCollectorWithBreakage( - t testing.TB, - env *ledgertestutils.IntegrationEnv, - breakageService ledgerbreakage.Service, -) *accrualCollector { - collector := newTestAccrualCollector(t, env) - collector.breakage = breakageService - collector.advance = advancetestutils.NewService(t, env.Deps, breakageService) - collector.transactionManager = enttx.NewCreator(env.DB) +func newTestAccrualCollector(t testing.TB, env *ledgertestutils.IntegrationEnv) *accrualCollector { + t.Helper() + breakageService := newTestBreakageService(t, env) + service, err := fbo.NewService(fbo.Config{ + Logger: omtestutils.NewDiscardLogger(t), + Advance: advancetestutils.NewService(t, env.Deps, breakageService), + Breakage: breakageService, + Dependencies: transactions.ResolverDependencies{ + AccountService: env.Deps.ResolversService, + AccountCatalog: env.Deps.AccountService, + BalanceQuerier: env.Deps.HistoricalLedger, + }, + AccountLocker: env.Deps.AccountService, + }) + require.NoError(t, err) - return collector + return &accrualCollector{ + ledger: env.Deps.HistoricalLedger, + fbo: service, + transactionManager: enttx.NewCreator(env.DB), + } } -func newTestBreakageService(t *testing.T, env *ledgertestutils.IntegrationEnv) ledgerbreakage.Service { +func newTestBreakageService(t testing.TB, env *ledgertestutils.IntegrationEnv) ledgerbreakage.Service { t.Helper() breakageAdapter, err := ledgerbreakageadapter.New(ledgerbreakageadapter.Config{ @@ -1027,7 +1035,7 @@ func TestCollectCustomerFBOMatchesPlanAndFeatureFilters(t *testing.T) { require.NoError(t, err) corrections, err := realized.CreateCorrectionRequest(alpacadecimal.NewFromInt(-35), fiat) require.NoError(t, err) - _, err = newTestAccrualCorrector(t, env, collector.breakage).Correct(t.Context(), CorrectCollectedAccruedInput{ + _, err = newTestAccrualCorrector(t, env).Correct(t.Context(), CorrectCollectedAccruedInput{ Namespace: env.Namespace, ChargeID: spendChargeID, CustomerID: env.CustomerID.ID, AllocateAt: env.Now(), Corrections: corrections, }) require.NoError(t, err) diff --git a/openmeter/ledger/collector/correct_test.go b/openmeter/ledger/collector/correct_test.go index d624286aa2..0834f69bf6 100644 --- a/openmeter/ledger/collector/correct_test.go +++ b/openmeter/ledger/collector/correct_test.go @@ -31,7 +31,7 @@ import ( func TestCollectToReceivableAndCorrectPreservesChargeProvenance(t *testing.T) { env := ledgertestutils.NewIntegrationEnv(t, "collector-receivable") collector := newTestAccrualCollector(t, env) - corrector := newTestAccrualCorrector(t, env, ledgerbreakage.NewNoopService()) + corrector := newTestAccrualCorrector(t, env) // given: settlement-fiat credit purchased by one charge sourceChargeID := testChargeID(1) @@ -103,7 +103,7 @@ func TestCollectToReceivableAndCorrectPreservesChargeProvenance(t *testing.T) { func TestCorrectCollectedAccruedUsesReverseFeatureAwareCollectionOrder(t *testing.T) { env := ledgertestutils.NewIntegrationEnv(t, "collector-correct") collector := newTestAccrualCollector(t, env) - corrector := newTestAccrualCorrector(t, env, ledgerbreakage.NewNoopService()) + corrector := newTestAccrualCorrector(t, env) // given: // - one spend charge fully consumes a feature-restricted source and an unrestricted source @@ -168,8 +168,8 @@ func TestCorrectCollectedAccruedUsesReverseFeatureAwareCollectionOrder(t *testin func TestCorrectCollectedAccruedReopensBreakageByReverseFeatureAwareCollectionOrder(t *testing.T) { env := ledgertestutils.NewIntegrationEnv(t, "collector-correct-breakage") breakageService := newTestBreakageService(t, env) - collector := newTestAccrualCollectorWithBreakage(t, env, breakageService) - corrector := newTestAccrualCorrector(t, env, breakageService) + collector := newTestAccrualCollector(t, env) + corrector := newTestAccrualCorrector(t, env) // given: // - two expiring sources are fully consumed and their planned breakage is released @@ -264,8 +264,8 @@ func TestCorrectCollectedAccruedReopensBreakageByReverseFeatureAwareCollectionOr func TestCorrectCollectedAccruedBreakageReopenTracksSourceOnBreakage(t *testing.T) { env := ledgertestutils.NewIntegrationEnv(t, "collector-correct-breakage") breakageService := newTestBreakageService(t, env) - collector := newTestAccrualCollectorWithBreakage(t, env, breakageService) - corrector := newTestAccrualCorrector(t, env, breakageService) + collector := newTestAccrualCollector(t, env) + corrector := newTestAccrualCorrector(t, env) // given: // - sourced expiring credit has been consumed by one spend charge @@ -345,7 +345,7 @@ func TestCorrectCollectedAccruedBreakageReopenTracksSourceOnBreakage(t *testing. func TestCorrectCollectedAccruedPreservesSourceAndSpendBuckets(t *testing.T) { env := ledgertestutils.NewIntegrationEnv(t, "collector-correct-provenance") collector := newTestAccrualCollector(t, env) - corrector := newTestAccrualCorrector(t, env, ledgerbreakage.NewNoopService()) + corrector := newTestAccrualCorrector(t, env) // given: // - one spend charge consumes two same-route source charges @@ -403,7 +403,7 @@ func TestCorrectCollectedAccruedPreservesSourceAndSpendBuckets(t *testing.T) { func TestCorrectCollectedAccruedPartiallyReversesAdvanceBackedCollection(t *testing.T) { env := ledgertestutils.NewIntegrationEnv(t, "collector-correct-advance") collector := newTestAccrualCollector(t, env) - corrector := newTestAccrualCorrector(t, env, ledgerbreakage.NewNoopService()) + corrector := newTestAccrualCorrector(t, env) // given: // - credit-only usage has no real source, so it creates advance receivable and accrued exposure @@ -471,7 +471,7 @@ func TestCorrectSourceLessCustomCurrencyPromotionalCollection(t *testing.T) { env := ledgertestutils.NewIntegrationEnv(t, "collector-correct-custom-promotion") env.Currency = currencyx.Code("ACME") collector := newTestAccrualCollector(t, env) - corrector := newTestAccrualCorrector(t, env, ledgerbreakage.NewNoopService()) + corrector := newTestAccrualCorrector(t, env) // given: // - 100 promotional ACME credits with no fiat exchange source @@ -583,7 +583,7 @@ func TestCorrectFiatFundedCustomCurrencyCreditOnlyShortfall(t *testing.T) { env := ledgertestutils.NewIntegrationEnv(t, "collector-correct-funded-custom-shortfall") env.Currency = currencyx.Code("ACME") collector := newTestAccrualCollector(t, env) - corrector := newTestAccrualCorrector(t, env, ledgerbreakage.NewNoopService()) + corrector := newTestAccrualCorrector(t, env) // given: // - 100 ACME credits funded from 25 USD at a copied 0.25 cost basis @@ -852,7 +852,7 @@ func TestCorrectCustomCurrencyCreditOnlyShortfall_MultipleFundingSourcesDoNotCro env := ledgertestutils.NewIntegrationEnv(t, "collector-correct-custom-multi-source") env.Currency = currencyx.Code("ACME") collector := newTestAccrualCollector(t, env) - corrector := newTestAccrualCorrector(t, env, ledgerbreakage.NewNoopService()) + corrector := newTestAccrualCorrector(t, env) // given: // - 100 ACME funded from USD at priority 1 (cost basis 0.25) @@ -1016,10 +1016,10 @@ func TestCorrectCustomCurrencyCreditOnlyShortfall_MultipleFundingSourcesDoNotCro func newTestAccrualCorrector( t testing.TB, env *ledgertestutils.IntegrationEnv, - breakageService ledgerbreakage.Service, ) *correction.Corrector { t.Helper() + breakageService := newTestBreakageService(t, env) corrector, err := correction.New(correction.Config{ Logger: omtestutils.NewDiscardLogger(t), Advance: advancetestutils.NewService(t, env.Deps, breakageService), @@ -1069,7 +1069,7 @@ func realizationsFromAllocations(env *ledgertestutils.IntegrationEnv, allocation func TestCorrectCollectedAccruedResumesCollapsedSourceSuffix(t *testing.T) { env := ledgertestutils.NewIntegrationEnv(t, "collector-correct-source-suffix") collector := newTestAccrualCollector(t, env) - corrector := newTestAccrualCorrector(t, env, ledgerbreakage.NewNoopService()) + corrector := newTestAccrualCorrector(t, env) // given: one allocation collapses two source charges, followed by another route. sourceA, sourceB, sourceC, spend := testChargeID(1), testChargeID(2), testChargeID(3), testChargeID(4) @@ -1110,7 +1110,7 @@ func TestCorrectCollectedAccruedResumesCollapsedSourceSuffix(t *testing.T) { func TestCorrectRecognizedBackfillSelectsOriginalSpend(t *testing.T) { env := ledgertestutils.NewIntegrationEnv(t, "collector-correct-shared-backfill") env.Currency = "ACME" - corrector := newTestAccrualCorrector(t, env, ledgerbreakage.NewNoopService()) + corrector := newTestAccrualCorrector(t, env) deps := transactions.ResolverDependencies{ AccountService: env.Deps.ResolversService, AccountCatalog: env.Deps.AccountService, diff --git a/openmeter/ledger/collector/service.go b/openmeter/ledger/collector/service.go index 5145ea8766..95cee20055 100644 --- a/openmeter/ledger/collector/service.go +++ b/openmeter/ledger/collector/service.go @@ -16,6 +16,7 @@ import ( "github.com/openmeterio/openmeter/openmeter/ledger/advance" "github.com/openmeterio/openmeter/openmeter/ledger/breakage" "github.com/openmeterio/openmeter/openmeter/ledger/collector/correction" + "github.com/openmeterio/openmeter/openmeter/ledger/fbo" "github.com/openmeterio/openmeter/openmeter/ledger/transactions" "github.com/openmeterio/openmeter/openmeter/productcatalog" "github.com/openmeterio/openmeter/pkg/framework/transaction" @@ -217,13 +218,21 @@ func NewService(config Config) (Service, error) { return nil, fmt.Errorf("create correction service: %w", err) } + fboService, err := fbo.NewService(fbo.Config{ + Logger: config.Logger, + Advance: config.Advance, + Dependencies: config.Dependencies, + Breakage: config.Breakage, + AccountLocker: config.AccountLocker, + }) + if err != nil { + return nil, fmt.Errorf("create FBO service: %w", err) + } + return &service{ collector: &accrualCollector{ ledger: config.Ledger, - advance: config.Advance, - deps: config.Dependencies, - breakage: config.Breakage, - accountLocker: config.AccountLocker, + fbo: fboService, transactionManager: config.TransactionManager, }, corrector: corrector, diff --git a/openmeter/ledger/fbo/README.md b/openmeter/ledger/fbo/README.md new file mode 100644 index 0000000000..ff13dc1ebf --- /dev/null +++ b/openmeter/ledger/fbo/README.md @@ -0,0 +1,21 @@ +# FBO consumption + +This package applies customer credit to usage or outstanding receivables. It +selects credit sources, releases their scheduled breakage as they are consumed, +and coordinates [advances](../advance/README.md). Billing decides the amount and +settlement policy; the collector commits the transactions and creates allocations. + +`Consume` returns resolved transaction inputs for one collection request. The +caller supplies the group ID and keeps source selection, breakage bookkeeping, +and group posting in one database transaction. Breakage releases reference +consumption entry IDs assigned before posting. Any failure requires rollback. + +Sources are selected by priority, restricted before unrestricted credit, earliest +expiry, then stable source cursor. `SourceBalanceAsOf` is the source-selection +cutoff; `BookedAt` is the effective time of consumption. Receivable coverage never +creates advance. + +`GetSpendableBalanceBuckets` reports positive capacity available at `AsOf` without +reducing any later recorded balance below zero. It takes the minimum per source +before combining buckets. Future breakage is excluded because consumption +releases it; breakage at or before `AsOf` remains part of the opening balance. diff --git a/openmeter/ledger/fbo/balance.go b/openmeter/ledger/fbo/balance.go new file mode 100644 index 0000000000..74fb67ac58 --- /dev/null +++ b/openmeter/ledger/fbo/balance.go @@ -0,0 +1,105 @@ +package fbo + +import ( + "context" + "errors" + "fmt" + "slices" + + "github.com/openmeterio/openmeter/openmeter/ledger" + "github.com/openmeterio/openmeter/pkg/models" +) + +// SpendableBalanceBucketQuery selects complete FBO sources. GroupBy may contain +// source_charge_id; omitting it sums independently spendable sources per subaccount. +// AsOf is the time at which credit must be available. +type SpendableBalanceBucketQuery ledger.BalanceBucketQuery + +var _ models.Validator = SpendableBalanceBucketQuery{} + +func (q SpendableBalanceBucketQuery) Validate() error { + var errs []error + if err := (ledger.MinimumBalanceBucketQuery{BalanceBucketQuery: ledger.BalanceBucketQuery(q)}).Validate(); err != nil { + errs = append(errs, err) + } + + if q.Filters.AccountID == nil || *q.Filters.AccountID == "" { + errs = append(errs, errors.New("FBO account id is required")) + } + + if q.Filters.TransactionID != nil || q.Filters.Provenance.SpendChargeID.IsPresent() || q.Filters.Provenance.CollectionOriginID.IsPresent() || len(q.ExcludeAnnotationFilters) > 0 { + errs = append(errs, errors.New("spendable balance requires complete source histories")) + } + + for _, dimension := range q.GroupBy { + if dimension != ledger.BalanceBucketGroupBySourceChargeID { + errs = append(errs, fmt.Errorf("unsupported spendable balance grouping: %s", dimension)) + } + } + + return models.NewNillableGenericValidationError(errors.Join(errs...)) +} + +// GetSpendableBalanceBuckets preserves recorded future spending. Only future +// breakage is excluded: consuming credit releases that scheduled breakage in the +// same group. Breakage at or before AsOf still reduces the opening balance. +func (s *service) GetSpendableBalanceBuckets(ctx context.Context, query SpendableBalanceBucketQuery) ([]ledger.BalanceBucket, error) { + if err := query.Validate(); err != nil { + return nil, err + } + + account, err := s.dependencies.AccountCatalog.GetAccountByID(ctx, models.NamespacedID{ + Namespace: query.Namespace, + ID: *query.Filters.AccountID, + }) + if err != nil { + return nil, err + } + + if account.Type() != ledger.AccountTypeCustomerFBO { + return nil, models.NewGenericValidationError(errors.New("spendable balance requires a customer FBO account")) + } + + perSource := ledger.BalanceBucketQuery(query) + perSource.GroupBy = []string{ledger.BalanceBucketGroupBySourceChargeID} + buckets, err := s.dependencies.BalanceQuerier.GetMinimumBalanceBuckets(ctx, ledger.MinimumBalanceBucketQuery{ + BalanceBucketQuery: perSource, + ExcludeFutureAnnotationFilters: map[string]string{ + ledger.AnnotationCollectionType: ledger.CollectionTypeBreakage, + }, + }) + if err != nil { + return nil, fmt.Errorf("query spendable FBO balance: %w", err) + } + + // Collection cannot borrow capacity between sources. Clamp each source before + // aggregating, rather than taking the minimum of their combined journal. + spendable := make([]ledger.BalanceBucket, 0, len(buckets)) + bySubaccount := make(map[string]int) + for _, bucket := range buckets { + if !bucket.SettledAmount.IsPositive() { + continue + } + + if slices.Contains(query.GroupBy, ledger.BalanceBucketGroupBySourceChargeID) { + spendable = append(spendable, bucket) + continue + } + + id := bucket.Address.SubAccountID() + if idx, ok := bySubaccount[id]; ok { + combined := &spendable[idx] + combined.SettledAmount = combined.SettledAmount.Add(bucket.SettledAmount) + combined.PendingAmount = combined.PendingAmount.Add(bucket.PendingAmount) + if bucket.OldestMatchingEntryCreatedAt.Before(combined.OldestMatchingEntryCreatedAt) { + combined.OldestMatchingEntryCreatedAt = bucket.OldestMatchingEntryCreatedAt + } + } else { + bucket.GroupByValues = map[string]*string{} + bySubaccount[id] = len(spendable) + spendable = append(spendable, bucket) + } + } + + return spendable, nil +} diff --git a/openmeter/ledger/fbo/balance_test.go b/openmeter/ledger/fbo/balance_test.go new file mode 100644 index 0000000000..1ef3ddb194 --- /dev/null +++ b/openmeter/ledger/fbo/balance_test.go @@ -0,0 +1,247 @@ +package fbo_test + +import ( + "context" + "testing" + "time" + + "github.com/alpacahq/alpacadecimal" + "github.com/oklog/ulid/v2" + "github.com/samber/lo" + "github.com/samber/mo" + "github.com/stretchr/testify/require" + + enttx "github.com/openmeterio/openmeter/openmeter/ent/tx" + "github.com/openmeterio/openmeter/openmeter/ledger" + "github.com/openmeterio/openmeter/openmeter/ledger/fbo" + ledgertestutils "github.com/openmeterio/openmeter/openmeter/ledger/testutils" + "github.com/openmeterio/openmeter/openmeter/ledger/transactions" + "github.com/openmeterio/openmeter/pkg/framework/transaction" + "github.com/openmeterio/openmeter/pkg/timeutil" +) + +func TestConsumptionPreservesFutureSpending(t *testing.T) { + type grant struct { + at time.Duration + amount int64 + } + + for _, tc := range []struct { + name string + grants []grant + futureSpend int64 + expected float64 + }{ + {name: "later consumption keeps its allocation", grants: []grant{{0, 10}}, futureSpend: 7, expected: 3}, + {name: "later consumption exhausts credit", grants: []grant{{0, 10}}, futureSpend: 10, expected: 0}, + {name: "future grant cannot fund earlier consumption", grants: []grant{{2 * time.Hour, 10}}, expected: 0}, + {name: "later grant backs later spending", grants: []grant{{0, 4}, {2 * time.Hour, 6}}, futureSpend: 6, expected: 4}, + } { + t.Run(tc.name, func(t *testing.T) { + env, service, _ := newFBOEnv(t) + at := env.Now() + sourceID := ulid.Make().String() + + // given: grants and a future consumption are already committed. + for _, grant := range tc.grants { + bookFBOTransactions(t, env, transactions.IssueCustomerReceivableTemplate{ + At: at.Add(grant.at), Amount: alpacadecimal.NewFromInt(grant.amount), + Currency: env.CurrencyReference(), SourceChargeID: &sourceID, + }) + } + + if tc.futureSpend > 0 { + future := consumeInput(env, "future-usage", tc.futureSpend) + future.BookedAt = at.Add(3 * time.Hour) + future.SourceBalanceAsOf = future.BookedAt + consumeAndCommit(t, env, service, future) + } + + // when: another collection is booked before that consumption. + input := consumeInput(env, "earlier-usage", 10) + input.BookedAt = at.Add(time.Hour) + input.SourceBalanceAsOf = input.BookedAt + consumeAndCommit(t, env, service, input) + + // then: it collects only credit left after preserving future spending. + requireAccountAmount(t, env, env.CustomerAccounts.AccruedAccount, float64(tc.futureSpend)+tc.expected) + }) + } +} + +func TestSpendableBalanceRespectsBreakageBoundary(t *testing.T) { + for _, tc := range []struct { + name string + at time.Duration + expected float64 + }{ + {name: "before breakage", at: 24 * time.Hour, expected: 3}, + {name: "at breakage", at: 10 * 24 * time.Hour, expected: 0}, + {name: "after breakage", at: 11 * 24 * time.Hour, expected: 0}, + } { + t.Run(tc.name, func(t *testing.T) { + env, service, breakageService := newFBOEnv(t) + + // given: ten expiring credits, seven consumed before their breakage date. + bookExpiringCredit(t, env, breakageService, 10) + future := consumeInput(env, "existing-usage", 7) + future.BookedAt = env.Now().Add(5 * 24 * time.Hour) + future.SourceBalanceAsOf = future.BookedAt + consumeAndCommit(t, env, service, future) + + // when: consumption is requested before, at, or after breakage. + input := consumeInput(env, "new-usage", 10) + input.BookedAt = env.Now().Add(tc.at) + input.SourceBalanceAsOf = input.BookedAt + consumeAndCommit(t, env, service, input) + + // then: future breakage can be released, but expired credit stays unavailable. + requireAccountAmount(t, env, env.CustomerAccounts.AccruedAccount, 7+tc.expected) + requireAccountAmount(t, env, env.CustomerAccounts.FBOAccount, 0) + }) + } +} + +func TestSpendableBalanceSumsIndependentSourceMinima(t *testing.T) { + env, service, _ := newFBOEnv(t) + at := env.Now() + sourceA, sourceB := ulid.Make().String(), ulid.Make().String() + + // given: source A dips to two, then is restored as source B dips to three. + // The combined journal never drops below twelve; collectable capacity is five. + for _, source := range []string{sourceA, sourceB} { + bookFBOTransactions(t, env, transactions.IssueCustomerReceivableTemplate{ + At: at, Amount: alpacadecimal.NewFromInt(10), Currency: env.CurrencyReference(), SourceChargeID: &source, + }) + } + + first := consumeInput(env, "first", 8) + first.BookedAt = at.Add(time.Hour) + first.SourceBalanceAsOf = first.BookedAt + original := consumeAndCommit(t, env, service, first) + require.Len(t, original.Transactions(), 1) + tx := original.Transactions()[0] + var negative, positive ledger.Entry + for _, entry := range tx.Entries() { + if entry.Amount().IsNegative() { + negative = entry + } else { + positive = entry + } + } + + require.NotNil(t, negative) + require.NotNil(t, positive) + require.Equal(t, sourceA, *negative.Provenance().SourceChargeID) + reversal, err := transactions.ReverseOriginEntryPair(transactions.ReverseOriginEntryPairInput{ + At: at.Add(2 * time.Hour), Amount: alpacadecimal.NewFromInt(8), Transaction: tx, + NegativeEntry: negative, PositiveEntry: positive, + }) + require.NoError(t, err) + _, err = env.Deps.HistoricalLedger.CommitGroup(t.Context(), reversal.AsGroupInput(env.Namespace, nil)) + require.NoError(t, err) + bookFBOTransactions(t, env, transactions.TransferCustomerFBOToAccruedTemplate{ + At: at.Add(2 * time.Hour), Currency: env.CurrencyReference(), + Sources: []transactions.PostingAmount{{ + Address: negative.PostingAddress(), Amount: alpacadecimal.NewFromInt(7), + Identity: ledger.EntryIdentityParts{Provenance: ledger.Provenance{SourceChargeID: &sourceB, SpendChargeID: lo.ToPtr("second")}}, + }}, + }) + + // when: the API omits source grouping at the opening boundary. + query := fbo.SpendableBalanceBucketQuery{Namespace: env.Namespace, Filters: ledger.Filters{ + AccountID: lo.ToPtr(env.CustomerAccounts.FBOAccount.ID().ID), AsOf: &at, + Route: ledger.RouteFilter{Currency: env.CurrencyReference()}, + }} + combined, err := service.GetSpendableBalanceBuckets(t.Context(), query) + require.NoError(t, err) + + // then: presentation grouping cannot borrow capacity between sources. + require.Len(t, combined, 1) + require.Empty(t, combined[0].GroupByValues) + require.Equal(t, 5.0, combined[0].SettledAmount.InexactFloat64()) + accountBalance, err := env.Deps.HistoricalLedger.GetAccountBalance(t.Context(), env.CustomerAccounts.FBOAccount, + query.Filters.Route, ledger.BalanceQuery{AsOf: &at}) + require.NoError(t, err) + require.Equal(t, 20.0, accountBalance.InexactFloat64()) + combinedMinimum, err := env.Deps.HistoricalLedger.GetMinimumBalanceBuckets(t.Context(), ledger.MinimumBalanceBucketQuery{ + BalanceBucketQuery: ledger.BalanceBucketQuery(query), + }) + require.NoError(t, err) + require.Len(t, combinedMinimum, 1) + require.Equal(t, 12.0, combinedMinimum[0].SettledAmount.InexactFloat64()) + + query.GroupBy = []string{ledger.BalanceBucketGroupBySourceChargeID} + bySource, err := service.GetSpendableBalanceBuckets(t.Context(), query) + require.NoError(t, err) + require.Len(t, bySource, 2) + require.Equal(t, 2.0, bySource[0].SettledAmount.InexactFloat64()) + require.Equal(t, 3.0, bySource[1].SettledAmount.InexactFloat64()) + + query.Filters.Provenance.SourceChargeID = mo.Some(&sourceB) + bySource, err = service.GetSpendableBalanceBuckets(t.Context(), query) + require.NoError(t, err) + require.Len(t, bySource, 1) + require.Equal(t, sourceB, *bySource[0].GroupByValues[ledger.BalanceBucketGroupBySourceChargeID]) + require.Equal(t, 3.0, bySource[0].SettledAmount.InexactFloat64()) + + // The committed return is available once its own effective boundary is reached. + query.Filters.AsOf = lo.ToPtr(at.Add(2 * time.Hour)) + query.Filters.Provenance.SourceChargeID = mo.Some(&sourceA) + returned, err := service.GetSpendableBalanceBuckets(t.Context(), query) + require.NoError(t, err) + require.Len(t, returned, 1) + require.Equal(t, 10.0, returned[0].SettledAmount.InexactFloat64()) + + query.Filters.AccountID = lo.ToPtr(env.CustomerAccounts.ReceivableAccount.ID().ID) + _, err = service.GetSpendableBalanceBuckets(t.Context(), query) + require.ErrorContains(t, err, "customer FBO account") +} + +func TestSpendableBalanceQueryRequiresCompleteSources(t *testing.T) { + at := time.Date(2026, time.January, 10, 0, 0, 0, 0, time.UTC) + for _, tc := range []struct { + name string + filters ledger.Filters + groupBy []string + }{ + {name: "missing boundary", filters: ledger.Filters{AccountID: lo.ToPtr("account")}}, + {name: "missing account", filters: ledger.Filters{AsOf: &at}}, + {name: "transaction", filters: ledger.Filters{AsOf: &at, AccountID: lo.ToPtr("account"), TransactionID: lo.ToPtr("transaction")}}, + {name: "period", filters: ledger.Filters{AsOf: &at, AccountID: lo.ToPtr("account"), BookedAtPeriod: &timeutil.OpenPeriod{From: &at}}}, + {name: "spend filter", filters: ledger.Filters{AsOf: &at, AccountID: lo.ToPtr("account"), Provenance: ledger.ProvenanceFilter{SpendChargeID: mo.Some(lo.ToPtr("spend"))}}}, + {name: "origin filter", filters: ledger.Filters{AsOf: &at, AccountID: lo.ToPtr("account"), Provenance: ledger.ProvenanceFilter{CollectionOriginID: mo.Some[*string](nil)}}}, + {name: "spend grouping", filters: ledger.Filters{AsOf: &at, AccountID: lo.ToPtr("account")}, groupBy: []string{ledger.BalanceBucketGroupBySpendChargeID}}, + } { + t.Run(tc.name, func(t *testing.T) { + require.Error(t, (fbo.SpendableBalanceBucketQuery{Namespace: "ns", Filters: tc.filters, GroupBy: tc.groupBy}).Validate()) + }) + } +} + +func bookFBOTransactions(t *testing.T, env *ledgertestutils.IntegrationEnv, templates ...transactions.TransactionTemplate) ledger.TransactionGroup { + t.Helper() + inputs, err := transactions.ResolveTransactions(t.Context(), transactions.ResolverDependencies{ + AccountService: env.Deps.ResolversService, AccountCatalog: env.Deps.AccountService, BalanceQuerier: env.Deps.HistoricalLedger, + }, transactions.ResolutionScope{CustomerID: env.CustomerID, Namespace: env.Namespace}, templates...) + require.NoError(t, err) + group, err := env.Deps.HistoricalLedger.CommitGroup(t.Context(), transactions.GroupInputs(env.Namespace, nil, inputs...)) + require.NoError(t, err) + + return group +} + +func consumeAndCommit(t *testing.T, env *ledgertestutils.IntegrationEnv, service fbo.Service, input fbo.ConsumeInput) ledger.TransactionGroup { + t.Helper() + group, err := transaction.Run(t.Context(), enttx.NewCreator(env.DB), func(ctx context.Context) (ledger.TransactionGroup, error) { + inputs, err := service.Consume(ctx, input) + if err != nil || len(inputs) == 0 { + return nil, err + } + + return env.Deps.HistoricalLedger.CommitGroup(ctx, ledger.WithGroupID(transactions.GroupInputs(env.Namespace, input.Annotations, inputs...), input.TransactionGroupID)) + }) + require.NoError(t, err) + + return group +} diff --git a/openmeter/ledger/fbo/consume.go b/openmeter/ledger/fbo/consume.go new file mode 100644 index 0000000000..4663c4e76b --- /dev/null +++ b/openmeter/ledger/fbo/consume.go @@ -0,0 +1,373 @@ +package fbo + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/alpacahq/alpacadecimal" + "github.com/oklog/ulid/v2" + "github.com/samber/lo" + + "github.com/openmeterio/openmeter/openmeter/currencies" + "github.com/openmeterio/openmeter/openmeter/customer" + "github.com/openmeterio/openmeter/openmeter/ledger" + "github.com/openmeterio/openmeter/openmeter/ledger/advance" + "github.com/openmeterio/openmeter/openmeter/ledger/breakage" + "github.com/openmeterio/openmeter/openmeter/ledger/transactions" + "github.com/openmeterio/openmeter/pkg/models" +) + +type ConsumeDestination string + +const ( + ConsumeDestinationAccrued ConsumeDestination = "accrued" + ConsumeDestinationReceivable ConsumeDestination = "receivable" +) + +var _ models.Validator = ConsumeDestination("") + +func (d ConsumeDestination) Validate() error { + switch d { + case ConsumeDestinationAccrued, ConsumeDestinationReceivable: + return nil + default: + return models.NewGenericValidationError(fmt.Errorf("invalid consume destination: %q", d)) + } +} + +type AdvancePolicy string + +const ( + AdvancePolicyNever AdvancePolicy = "never" + AdvancePolicyOnShortfall AdvancePolicy = "on_shortfall" +) + +var _ models.Validator = AdvancePolicy("") + +func (p AdvancePolicy) Validate() error { + switch p { + case AdvancePolicyNever, AdvancePolicyOnShortfall: + return nil + default: + return models.NewGenericValidationError(fmt.Errorf("invalid advance policy: %q", p)) + } +} + +type ConsumeInput struct { + breakage.PostingInput + CustomerID customer.CustomerID + ChargeID string + BookedAt time.Time + SourceBalanceAsOf time.Time + Currency currencies.CurrencyReference + Filters ledger.CreditFilters + Amount alpacadecimal.Decimal + Destination ConsumeDestination + AdvancePolicy AdvancePolicy + TaxCode *string + TaxBehavior *ledger.TaxBehavior +} + +var _ models.Validator = ConsumeInput{} + +func (i ConsumeInput) Validate() error { + var errs []error + if err := i.CustomerID.Validate(); err != nil { + errs = append(errs, fmt.Errorf("customer: %w", err)) + } + + if err := i.PostingInput.Validate(); err != nil { + errs = append(errs, fmt.Errorf("posting: %w", err)) + } + + if i.ChargeID == "" { + errs = append(errs, errors.New("charge id is required")) + } + + if i.BookedAt.IsZero() { + errs = append(errs, errors.New("booked at is required")) + } + + if i.SourceBalanceAsOf.IsZero() { + errs = append(errs, errors.New("source balance as of is required")) + } + + if err := i.Currency.Validate(); err != nil { + errs = append(errs, fmt.Errorf("currency: %w", err)) + } else if i.Currency.IsCustom() && !i.Currency.IsResolved() { + errs = append(errs, errors.New("custom currency must be resolved")) + } + + if err := i.Filters.Validate(); err != nil { + errs = append(errs, fmt.Errorf("filters: %w", err)) + } + + if i.Amount.IsNegative() { + errs = append(errs, errors.New("amount cannot be negative")) + } else if i.Amount.IsPositive() { + if err := ledger.ValidateTransactionAmount(i.Amount); err != nil { + errs = append(errs, fmt.Errorf("amount: %w", err)) + } + } + + if err := i.Destination.Validate(); err != nil { + errs = append(errs, fmt.Errorf("destination: %w", err)) + } + + if err := i.AdvancePolicy.Validate(); err != nil { + errs = append(errs, fmt.Errorf("advance policy: %w", err)) + } + + if i.Destination == ConsumeDestinationReceivable { + if !i.Currency.IsFiat() { + errs = append(errs, errors.New("receivable coverage requires fiat currency")) + } + + if i.AdvancePolicy != AdvancePolicyNever { + errs = append(errs, errors.New("receivable coverage cannot create advance")) + } + + if i.TaxCode != nil || i.TaxBehavior != nil { + errs = append(errs, errors.New("receivable coverage cannot set accrued tax dimensions")) + } + } + + if i.TaxBehavior != nil { + if err := i.TaxBehavior.Validate(); err != nil { + errs = append(errs, fmt.Errorf("tax behavior: %w", err)) + } + } + + return models.NewNillableGenericValidationError(errors.Join(errs...)) +} + +// Consume selects credit and returns the resolved consumption, breakage release, +// and shortfall advance transactions. The caller must commit them under the supplied +// group ID in the same database transaction: breakage bookkeeping is written here, +// so any later failure requires rollback. +func (s *service) Consume(ctx context.Context, request ConsumeInput) ([]ledger.TransactionInput, error) { + if err := request.Validate(); err != nil { + return nil, err + } + + if request.Amount.IsZero() { + return nil, nil + } + + selectedCredits, creditTransactions, err := s.selectCreditsAndBuildTransactions(ctx, request) + if err != nil { + return nil, err + } + + // Each origin identifies one concrete FBO debit. Assign its IDs before + // writing breakage records that reference the still-unposted consumption. + sourceDebitsByOrigin, err := assignSourceDebitIDs(creditTransactions) + if err != nil { + return nil, err + } + + breakageReleases, err := s.releaseConsumedBreakage(ctx, releaseConsumedBreakageInput{ + ReleasePosting: request.PostingInput, + SpendChargeID: request.ChargeID, + SelectedCredits: selectedCredits, + SourceDebitsByOrigin: sourceDebitsByOrigin, + }) + if err != nil { + return nil, err + } + + postings := append(creditTransactions, breakageReleases...) + shortfall := request.Amount + for _, credit := range selectedCredits { + shortfall = shortfall.Sub(credit.amount) + } + + // Advances fund only the shortfall; they are issued and consumed together. + if request.AdvancePolicy == AdvancePolicyOnShortfall && shortfall.IsPositive() { + advancePostings, err := s.advance.PlanIssue(ctx, advance.IssueInput{ + CustomerID: request.CustomerID, + ChargeID: request.ChargeID, + At: request.BookedAt, + Amount: shortfall, + Currency: request.Currency, + Filters: request.Filters, + TaxCode: request.TaxCode, + TaxBehavior: request.TaxBehavior, + }) + if err != nil { + return nil, fmt.Errorf("plan FBO shortfall advance: %w", err) + } + + postings = append(postings, advancePostings...) + } + + for idx, posting := range postings { + postings[idx] = transactions.WithAnnotations(posting, request.Annotations) + } + + return postings, nil +} + +// selectCreditsAndBuildTransactions applies source priority and builds +// transactions for the requested destination. Each selected +// slice gets its own collection origin, even when slices share an issuing charge. +func (s *service) selectCreditsAndBuildTransactions(ctx context.Context, request ConsumeInput) (selectionList, []ledger.TransactionInput, error) { + sources, err := s.listCustomerFBOSources(ctx, SourceQuery{ + CustomerID: request.CustomerID, + Currency: request.Currency, + Filters: request.Filters, + AsOf: request.SourceBalanceAsOf, + }) + if err != nil { + return nil, nil, err + } + + selectedCredits := selectFBOSources(sources, request.Amount) + for idx := range selectedCredits { + selectedCredits[idx].collectionOriginID = lo.ToPtr(ulid.Make().String()) + } + + sourceAmounts := selectionList(selectedCredits).postingAmounts(&request.ChargeID) + if len(sourceAmounts) == 0 { + return selectedCredits, nil, nil + } + + var template transactions.TransactionTemplate + switch request.Destination { + case ConsumeDestinationAccrued: + template = transactions.TransferCustomerFBOToAccruedTemplate{ + At: request.BookedAt, + Currency: request.Currency, + TaxCode: request.TaxCode, + TaxBehavior: request.TaxBehavior, + Sources: sourceAmounts, + } + case ConsumeDestinationReceivable: + template = transactions.CoverCustomerReceivableTemplate{ + At: request.BookedAt, + Currency: request.Currency, + Sources: sourceAmounts, + } + } + + creditTransactions, err := transactions.ResolveTransactions(ctx, s.dependencies, transactions.ResolutionScope{ + CustomerID: request.CustomerID, + Namespace: request.CustomerID.Namespace, + }, template) + if err != nil { + return nil, nil, fmt.Errorf("resolve FBO consumption: %w", err) + } + + return selectedCredits, creditTransactions, nil +} + +type sourcePostingIDs struct { + TransactionID string + EntryID string +} + +// assignSourceDebitIDs replaces transactions with identified copies and returns the +// exact FBO debit references used to link breakage releases to consumed credit. +func assignSourceDebitIDs(creditTransactions []ledger.TransactionInput) (map[string]sourcePostingIDs, error) { + sourceDebitsByOrigin := make(map[string]sourcePostingIDs, len(creditTransactions)) + for idx, posting := range creditTransactions { + identified, err := ledger.PreassignIDs(posting) + if err != nil { + return nil, err + } + + creditTransactions[idx] = identified + for _, entry := range identified.EntryInputs() { + if entry.PostingAddress().AccountType() != ledger.AccountTypeCustomerFBO || !entry.Amount().IsNegative() { + continue + } + + originID := lo.FromPtr(entry.Provenance().CollectionOriginID) + if _, exists := sourceDebitsByOrigin[originID]; exists { + return nil, fmt.Errorf("collection origin %s has multiple FBO source entries", originID) + } + + sourceDebitsByOrigin[originID] = sourcePostingIDs{ + TransactionID: identified.AssignedID(), + EntryID: entry.AssignedID(), + } + } + } + + return sourceDebitsByOrigin, nil +} + +type releaseConsumedBreakageInput struct { + ReleasePosting breakage.PostingInput + SpendChargeID string + SelectedCredits selectionList + SourceDebitsByOrigin map[string]sourcePostingIDs +} + +var _ models.Validator = releaseConsumedBreakageInput{} + +func (i releaseConsumedBreakageInput) Validate() error { + var errs []error + if err := i.ReleasePosting.Validate(); err != nil { + errs = append(errs, err) + } + + if i.SpendChargeID == "" { + errs = append(errs, errors.New("charge id is required")) + } + + return models.NewNillableGenericValidationError(errors.Join(errs...)) +} + +// releaseConsumedBreakage writes release records for selected expiring credit. +// Their transactions join the same group as the source debits they reference. +func (s *service) releaseConsumedBreakage(ctx context.Context, input releaseConsumedBreakageInput) ([]ledger.TransactionInput, error) { + if err := input.Validate(); err != nil { + return nil, err + } + + var breakageReleases []ledger.TransactionInput + remainingBreakageByPlanID := make(map[string]alpacadecimal.Decimal) + for _, selection := range input.SelectedCredits { + consumedSource, ok := input.SourceDebitsByOrigin[lo.FromPtr(selection.collectionOriginID)] + if !ok { + return nil, errors.New("selected credit has no resolved FBO source posting") + } + + if selection.source.breakagePlan != nil { + breakagePlan := *selection.source.breakagePlan + remaining, ok := remainingBreakageByPlanID[breakagePlan.ID.ID] + if !ok { + remaining = breakagePlan.OpenAmount + } + + // Legacy source-less plans can span several selected source slices; + // cap their combined release before writing more bookkeeping. + if selection.amount.GreaterThan(remaining) { + return nil, fmt.Errorf("release amount exceeds remaining breakage plan amount for %s", breakagePlan.ID.ID) + } + + remainingBreakageByPlanID[breakagePlan.ID.ID] = remaining.Sub(selection.amount) + breakageRelease, err := s.breakage.ReleasePlan(ctx, breakage.ReleasePlanInput{ + PostingInput: input.ReleasePosting, + Plan: breakagePlan, + Amount: selection.amount, + SourceKind: breakage.SourceKindUsage, + SourceChargeID: selection.source.sourceChargeID, + SpendChargeID: &input.SpendChargeID, + CollectionOriginID: selection.collectionOriginID, + SourceTransactionID: &consumedSource.TransactionID, + SourceEntryID: &consumedSource.EntryID, + }) + if err != nil { + return nil, fmt.Errorf("release consumed credit breakage: %w", err) + } + + breakageReleases = append(breakageReleases, breakageRelease) + } + } + + return breakageReleases, nil +} diff --git a/openmeter/ledger/fbo/consume_test.go b/openmeter/ledger/fbo/consume_test.go new file mode 100644 index 0000000000..07c929fae1 --- /dev/null +++ b/openmeter/ledger/fbo/consume_test.go @@ -0,0 +1,252 @@ +package fbo_test + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/alpacahq/alpacadecimal" + "github.com/oklog/ulid/v2" + "github.com/samber/lo" + "github.com/stretchr/testify/require" + + currencytestutils "github.com/openmeterio/openmeter/openmeter/currencies/testutils" + enttx "github.com/openmeterio/openmeter/openmeter/ent/tx" + "github.com/openmeterio/openmeter/openmeter/ledger" + advancetestutils "github.com/openmeterio/openmeter/openmeter/ledger/advance/testutils" + "github.com/openmeterio/openmeter/openmeter/ledger/breakage" + breakageadapter "github.com/openmeterio/openmeter/openmeter/ledger/breakage/adapter" + "github.com/openmeterio/openmeter/openmeter/ledger/fbo" + ledgertestutils "github.com/openmeterio/openmeter/openmeter/ledger/testutils" + "github.com/openmeterio/openmeter/openmeter/ledger/transactions" + omtestutils "github.com/openmeterio/openmeter/openmeter/testutils" + "github.com/openmeterio/openmeter/pkg/currencyx" + "github.com/openmeterio/openmeter/pkg/framework/transaction" +) + +func TestConsumptionKeepsSameCodeCurrencyBreakageSeparate(t *testing.T) { + env, service, breakageService := newFBOEnv(t) + env.Currency = currencyx.Code("ACME") + alpha := currencytestutils.NewCustomCurrency(t, env.Currency, 2) + beta := currencytestutils.NewCustomCurrency(t, env.Currency, 2) + + // given: only beta's credit expires, although both currencies have the same code. + env.CustomCurrency = &beta + bookExpiringCredit(t, env, breakageService, 10) + env.CustomCurrency = &alpha + fundCredit(t, env, 10) + + err := transaction.RunWithNoValue(t.Context(), enttx.NewCreator(env.DB), func(ctx context.Context) error { + // when: alpha is consumed through the FBO service. + input := consumeInput(env, "alpha-consumption", 6) + inputs, err := service.Consume(ctx, input) + require.NoError(t, err) + + // then: beta's breakage plan is neither attached nor released. + require.Len(t, inputs, 1) + plans, err := breakageService.ListPlans(ctx, breakage.ListPlansInput{ + CustomerID: env.CustomerID, + Currency: env.Currency, + AsOf: env.Now(), + }) + require.NoError(t, err) + require.Len(t, plans, 1) + require.Equal(t, 10.0, plans[0].OpenAmount.InexactFloat64()) + groupInput := ledger.WithGroupID(transactions.GroupInputs(env.Namespace, nil, inputs...), input.TransactionGroupID) + _, err = env.Deps.HistoricalLedger.CommitGroup(ctx, groupInput) + + return err + }) + require.NoError(t, err) + requireAccountAmount(t, env, env.CustomerAccounts.FBOAccount, 4) + env.CustomCurrency = &beta + balance, err := env.Deps.HistoricalLedger.GetAccountBalance(t.Context(), env.CustomerAccounts.FBOAccount, + ledger.RouteFilter{Currency: env.CurrencyReference()}, ledger.BalanceQuery{AsOf: lo.ToPtr(env.Now())}) + require.NoError(t, err) + require.Equal(t, 10.0, balance.InexactFloat64()) +} + +func TestConsumptionComposesBreakageAndAdvanceInOneGroup(t *testing.T) { + env, service, breakageService := newFBOEnv(t) + bookExpiringCredit(t, env, breakageService, 6) + + // given: expiring credit covers only part of credit-only consumption. + err := transaction.RunWithNoValue(t.Context(), enttx.NewCreator(env.DB), func(ctx context.Context) error { + input := consumeInput(env, "credit-only", 10) + input.AdvancePolicy = fbo.AdvancePolicyOnShortfall + + // when: consumption is planned without posting individual accounting legs. + inputs, err := service.Consume(ctx, input) + require.NoError(t, err) + require.Len(t, inputs, 4) + + var sourceEntry ledger.EntryInput + for _, entry := range inputs[0].EntryInputs() { + if entry.PostingAddress().AccountType() == ledger.AccountTypeCustomerFBO && entry.Amount().IsNegative() { + sourceEntry = entry + } + } + + require.NotNil(t, sourceEntry) + require.Equal(t, -6.0, sourceEntry.Amount().InexactFloat64()) + + releases, err := breakageService.ListReleases(ctx, breakage.ListReleasesInput{ + CustomerID: env.CustomerID, + SourceEntryID: []string{sourceEntry.AssignedID()}, + }) + require.NoError(t, err) + require.Len(t, releases, 1) + require.Equal(t, inputs[0].AssignedID(), *releases[0].SourceTransactionID) + require.Equal(t, input.TransactionGroupID, releases[0].BreakageTransactionGroupID) + + // then: consumption, release, and advance legs are committed in one caller-owned group. + groupInput := ledger.WithGroupID(transactions.GroupInputs(env.Namespace, nil, inputs...), input.TransactionGroupID) + group, err := env.Deps.HistoricalLedger.CommitGroup(ctx, groupInput) + require.NoError(t, err) + require.Len(t, group.Transactions(), 4) + + return nil + }) + require.NoError(t, err) + requireAccountAmount(t, env, env.CustomerAccounts.FBOAccount, 0) + requireAccountAmount(t, env, env.CustomerAccounts.AccruedAccount, 10) +} + +func TestConsumptionRollbackDiscardsJournalAndBreakageRelease(t *testing.T) { + env, service, breakageService := newFBOEnv(t) + bookExpiringCredit(t, env, breakageService, 10) + failure := errors.New("caller failed") + var groupID string + + // given: an expiring consume and its bookkeeping share the caller transaction. + err := transaction.RunWithNoValue(t.Context(), enttx.NewCreator(env.DB), func(ctx context.Context) error { + input := consumeInput(env, "rollback", 4) + groupID = input.TransactionGroupID + inputs, err := service.Consume(ctx, input) + require.NoError(t, err) + groupInput := ledger.WithGroupID(transactions.GroupInputs(env.Namespace, nil, inputs...), input.TransactionGroupID) + _, err = env.Deps.HistoricalLedger.CommitGroup(ctx, groupInput) + require.NoError(t, err) + + // when: the caller fails after the nested ledger commit. + return failure + }) + require.ErrorIs(t, err, failure) + + // then: neither the group nor its breakage bookkeeping survives. + count, err := env.DB.LedgerTransactionGroup.Query().Count(t.Context()) + require.NoError(t, err) + require.Equal(t, 1, count) + releases, err := breakageService.ListReleases(t.Context(), breakage.ListReleasesInput{ + CustomerID: env.CustomerID, + SourceTransactionGroupID: []string{groupID}, + }) + require.NoError(t, err) + require.Empty(t, releases) + plans, err := breakageService.ListPlans(t.Context(), breakage.ListPlansInput{ + CustomerID: env.CustomerID, + Currency: env.Currency, + AsOf: env.Now(), + }) + require.NoError(t, err) + require.Len(t, plans, 1) + require.Equal(t, 10.0, plans[0].OpenAmount.InexactFloat64()) +} + +func newFBOEnv(t *testing.T) (*ledgertestutils.IntegrationEnv, fbo.Service, breakage.Service) { + t.Helper() + env := ledgertestutils.NewIntegrationEnv(t, "fbo-consumption") + adapter, err := breakageadapter.New(breakageadapter.Config{Client: env.DB}) + require.NoError(t, err) + dependencies := transactions.ResolverDependencies{ + AccountService: env.Deps.ResolversService, + AccountCatalog: env.Deps.AccountService, + BalanceQuerier: env.Deps.HistoricalLedger, + } + breakageService, err := breakage.NewService(breakage.Config{Adapter: adapter, Dependencies: dependencies}) + require.NoError(t, err) + service, err := fbo.NewService(fbo.Config{ + Logger: omtestutils.NewDiscardLogger(t), + Dependencies: dependencies, + Advance: advancetestutils.NewService(t, env.Deps, breakageService), + Breakage: breakageService, + AccountLocker: env.Deps.AccountService, + }) + require.NoError(t, err) + + return env, service, breakageService +} + +func consumeInput(env *ledgertestutils.IntegrationEnv, chargeID string, amount int64) fbo.ConsumeInput { + return fbo.ConsumeInput{ + CustomerID: env.CustomerID, + PostingInput: breakage.PostingInput{TransactionGroupID: ulid.Make().String()}, + ChargeID: chargeID, + BookedAt: env.Now(), + SourceBalanceAsOf: env.Now(), + Currency: env.CurrencyReference(), + Amount: alpacadecimal.NewFromInt(amount), + Destination: fbo.ConsumeDestinationAccrued, + AdvancePolicy: fbo.AdvancePolicyNever, + } +} + +func fundCredit(t *testing.T, env *ledgertestutils.IntegrationEnv, amount int64) { + t.Helper() + inputs, err := transactions.ResolveTransactions(t.Context(), transactions.ResolverDependencies{ + AccountService: env.Deps.ResolversService, + AccountCatalog: env.Deps.AccountService, + BalanceQuerier: env.Deps.HistoricalLedger, + }, transactions.ResolutionScope{CustomerID: env.CustomerID, Namespace: env.Namespace}, transactions.IssueCustomerReceivableTemplate{ + At: env.Now(), + Amount: alpacadecimal.NewFromInt(amount), + Currency: env.CurrencyReference(), + }) + require.NoError(t, err) + _, err = env.Deps.HistoricalLedger.CommitGroup(t.Context(), transactions.GroupInputs(env.Namespace, nil, inputs...)) + require.NoError(t, err) +} + +func bookExpiringCredit(t *testing.T, env *ledgertestutils.IntegrationEnv, breakageService breakage.Service, amount int64) { + t.Helper() + groupID := ulid.Make().String() + err := transaction.RunWithNoValue(t.Context(), enttx.NewCreator(env.DB), func(ctx context.Context) error { + inputs, err := transactions.ResolveTransactions(ctx, transactions.ResolverDependencies{ + AccountService: env.Deps.ResolversService, + AccountCatalog: env.Deps.AccountService, + BalanceQuerier: env.Deps.HistoricalLedger, + }, transactions.ResolutionScope{CustomerID: env.CustomerID, Namespace: env.Namespace}, transactions.IssueCustomerReceivableTemplate{ + At: env.Now(), + Amount: alpacadecimal.NewFromInt(amount), + Currency: env.CurrencyReference(), + }) + if err != nil { + return err + } + + breakageInputs, err := breakageService.PlanIssuance(ctx, breakage.PlanIssuanceInput{ + PostingInput: breakage.PostingInput{TransactionGroupID: groupID}, + CustomerID: env.CustomerID, + Amount: alpacadecimal.NewFromInt(amount), + Currency: env.CurrencyReference(), + ExpiresAt: env.Now().Add(10 * 24 * time.Hour), + }) + if err != nil { + return err + } + + inputs = append(inputs, breakageInputs...) + _, err = env.Deps.HistoricalLedger.CommitGroup(ctx, ledger.WithGroupID(transactions.GroupInputs(env.Namespace, nil, inputs...), groupID)) + + return err + }) + require.NoError(t, err) +} + +func requireAccountAmount(t *testing.T, env *ledgertestutils.IntegrationEnv, account ledger.Account, expected float64) { + t.Helper() + balance, err := env.Deps.HistoricalLedger.GetAccountBalance(t.Context(), account, ledger.RouteFilter{Currency: env.CurrencyReference()}, ledger.BalanceQuery{}) + require.NoError(t, err) + require.Equal(t, expected, balance.InexactFloat64()) +} diff --git a/openmeter/ledger/fbo/service.go b/openmeter/ledger/fbo/service.go new file mode 100644 index 0000000000..ef83afe6e2 --- /dev/null +++ b/openmeter/ledger/fbo/service.go @@ -0,0 +1,84 @@ +package fbo + +import ( + "context" + "errors" + "log/slog" + + "github.com/openmeterio/openmeter/openmeter/ledger" + "github.com/openmeterio/openmeter/openmeter/ledger/advance" + "github.com/openmeterio/openmeter/openmeter/ledger/breakage" + "github.com/openmeterio/openmeter/openmeter/ledger/transactions" + "github.com/openmeterio/openmeter/pkg/models" +) + +type Service interface { + GetSpendableBalanceBuckets(ctx context.Context, query SpendableBalanceBucketQuery) ([]ledger.BalanceBucket, error) + Consume(ctx context.Context, input ConsumeInput) ([]ledger.TransactionInput, error) + ListSources(ctx context.Context, input SourceQuery) ([]AvailableSource, error) +} + +type Config struct { + Logger *slog.Logger + Dependencies transactions.ResolverDependencies + Advance advance.Service + Breakage breakage.Service + AccountLocker ledger.AccountLocker +} + +var _ models.Validator = Config{} + +func (c Config) Validate() error { + var errs []error + if c.Logger == nil { + errs = append(errs, errors.New("logger is required")) + } + + if c.Dependencies.AccountService == nil { + errs = append(errs, errors.New("account service is required")) + } + + if c.Dependencies.AccountCatalog == nil { + errs = append(errs, errors.New("account catalog is required")) + } + + if c.Dependencies.BalanceQuerier == nil { + errs = append(errs, errors.New("balance querier is required")) + } + + if c.Advance == nil { + errs = append(errs, errors.New("advance service is required")) + } + + if c.Breakage == nil { + errs = append(errs, errors.New("breakage service is required")) + } + + if c.AccountLocker == nil { + errs = append(errs, errors.New("account locker is required")) + } + + return models.NewNillableGenericValidationError(errors.Join(errs...)) +} + +type service struct { + dependencies transactions.ResolverDependencies + advance advance.Service + breakage breakage.Service + accountLocker ledger.AccountLocker +} + +var _ Service = (*service)(nil) + +func NewService(config Config) (Service, error) { + if err := config.Validate(); err != nil { + return nil, err + } + + return &service{ + dependencies: config.Dependencies, + advance: config.Advance, + breakage: config.Breakage, + accountLocker: config.AccountLocker, + }, nil +} diff --git a/openmeter/ledger/fbo/source_query.go b/openmeter/ledger/fbo/source_query.go new file mode 100644 index 0000000000..296e347ac9 --- /dev/null +++ b/openmeter/ledger/fbo/source_query.go @@ -0,0 +1,84 @@ +package fbo + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/alpacahq/alpacadecimal" + + "github.com/openmeterio/openmeter/openmeter/currencies" + "github.com/openmeterio/openmeter/openmeter/customer" + "github.com/openmeterio/openmeter/openmeter/ledger" + "github.com/openmeterio/openmeter/pkg/models" +) + +type SourceQuery struct { + CustomerID customer.CustomerID + Currency currencies.CurrencyReference + Filters ledger.CreditFilters + AsOf time.Time +} + +var _ models.Validator = SourceQuery{} + +func (q SourceQuery) Validate() error { + var errs []error + if err := q.CustomerID.Validate(); err != nil { + errs = append(errs, fmt.Errorf("customer: %w", err)) + } + + if err := q.Currency.Validate(); err != nil { + errs = append(errs, fmt.Errorf("currency: %w", err)) + } else if q.Currency.IsCustom() && !q.Currency.IsResolved() { + errs = append(errs, errors.New("custom currency must be resolved")) + } + + if err := q.Filters.Validate(); err != nil { + errs = append(errs, fmt.Errorf("filters: %w", err)) + } + + if q.AsOf.IsZero() { + errs = append(errs, errors.New("as of is required")) + } + + return models.NewNillableGenericValidationError(errors.Join(errs...)) +} + +type AvailableSource struct { + Address ledger.PostingAddress + SourceChargeID *string + Amount alpacadecimal.Decimal + ExpiresAt *time.Time + BreakagePlanID *string +} + +// ListSources returns ordered spendable source slices, with their breakage plans. +func (s *service) ListSources(ctx context.Context, query SourceQuery) ([]AvailableSource, error) { + if err := query.Validate(); err != nil { + return nil, err + } + + sources, err := s.listCustomerFBOSources(ctx, query) + if err != nil { + return nil, err + } + + out := make([]AvailableSource, 0, len(sources)) + for _, source := range sources { + available := AvailableSource{ + Address: source.address, + SourceChargeID: source.sourceChargeID, + Amount: source.available, + ExpiresAt: source.expiresAt, + } + if source.breakagePlan != nil { + available.BreakagePlanID = &source.breakagePlan.ID.ID + } + + out = append(out, available) + } + + return out, nil +} diff --git a/openmeter/ledger/collector/collection_fbo.go b/openmeter/ledger/fbo/sources.go similarity index 63% rename from openmeter/ledger/collector/collection_fbo.go rename to openmeter/ledger/fbo/sources.go index 2f24f962d1..dad6074aa3 100644 --- a/openmeter/ledger/collector/collection_fbo.go +++ b/openmeter/ledger/fbo/sources.go @@ -1,54 +1,25 @@ -package collector +package fbo import ( "context" "fmt" "slices" - "time" "github.com/alpacahq/alpacadecimal" "github.com/samber/lo" "github.com/samber/mo" "github.com/openmeterio/openmeter/openmeter/currencies" - "github.com/openmeterio/openmeter/openmeter/customer" "github.com/openmeterio/openmeter/openmeter/ledger" "github.com/openmeterio/openmeter/openmeter/ledger/breakage" "github.com/openmeterio/openmeter/pkg/cmpx" ) -// collectCustomerFBOSelections builds the sources used by FBO->accrued -// collection: -// - query live FBO balance at route + source-charge granularity -// - if breakage is enabled, attach open breakage plans to matching source slices -// - select up to target from the prioritized sources -func (c *accrualCollector) collectCustomerFBOSelections( +func (c *service) listCustomerFBOSources( ctx context.Context, - customerID customer.CustomerID, - currency currencies.CurrencyReference, - targetRoute ledger.Route, - target alpacadecimal.Decimal, - asOf time.Time, -) ([]fboCollectionSelection, error) { - sources, err := c.listCustomerFBOSources(ctx, customerID, currency, targetRoute, asOf) - if err != nil { - return nil, err - } - - // prioritize FBO sources before final collection. - slices.SortStableFunc(sources, cmpx.Compare[fboCollectionSource]) - - return selectFBOSources(sources, target), nil -} - -func (c *accrualCollector) listCustomerFBOSources( - ctx context.Context, - customerID customer.CustomerID, - currency currencies.CurrencyReference, - targetRoute ledger.Route, - asOf time.Time, -) ([]fboCollectionSource, error) { - customerAccounts, err := c.deps.AccountService.GetCustomerAccounts(ctx, customerID) + query SourceQuery, +) ([]source, error) { + customerAccounts, err := c.dependencies.AccountService.GetCustomerAccounts(ctx, query.CustomerID) if err != nil { return nil, fmt.Errorf("get customer accounts: %w", err) } @@ -59,47 +30,48 @@ func (c *accrualCollector) listCustomerFBOSources( sources, err := c.listCustomerFBOBalanceBucketSources( ctx, - customerID.Namespace, + query, customerAccounts.FBOAccount.ID().ID, - currency, - targetRoute, - asOf, ) if err != nil { return nil, err } // prioritize FBO sources before breakage reserves source balances. - slices.SortStableFunc(sources, cmpx.Compare[fboCollectionSource]) + slices.SortStableFunc(sources, cmpx.Compare[source]) + + sources, err = c.mapBreakagePlansToFBOCollectionSources(ctx, query, sources) + if err != nil { + return nil, err + } + + slices.SortStableFunc(sources, cmpx.Compare[source]) - return c.mapBreakagePlansToFBOCollectionSources(ctx, customerID, currency, targetRoute, asOf, sources) + return sources, nil } -func (c *accrualCollector) mapBreakagePlansToFBOCollectionSources( +func (c *service) mapBreakagePlansToFBOCollectionSources( ctx context.Context, - customerID customer.CustomerID, - currency currencies.CurrencyReference, - targetRoute ledger.Route, - asOf time.Time, - sources []fboCollectionSource, -) ([]fboCollectionSource, error) { + query SourceQuery, + sources []source, +) ([]source, error) { // Breakage plans decide which expiring credit is considered first. The FBO // balance buckets decide whether that plan still has live source balance // available to collect. openPlans, err := c.breakage.ListPlans(ctx, breakage.ListPlansInput{ - CustomerID: customerID, - Currency: currency.Code, - AsOf: asOf, + CustomerID: query.CustomerID, + Currency: query.Currency.Code, + AsOf: query.AsOf, }) if err != nil { return nil, fmt.Errorf("list open breakage plans: %w", err) } - breakageSources := make([]fboCollectionSource, 0, len(openPlans)+len(sources)) + breakageSources := make([]source, 0, len(openPlans)+len(sources)) for _, plan := range openPlans { - // ponytail: ListPlans filters by code; keep the exact managed identity + // ListPlans filters by code; keep the exact managed identity // check here until same-code plan locking warrants an adapter-level filter. - reservedSources := reserveSourcesForBreakagePlan(sources, plan, currency, targetRoute) + reservedSources := reserveSourcesForBreakagePlan(sources, plan, query.Currency, ledger.Route{Filters: query.Filters}) if len(reservedSources) == 0 { continue } @@ -108,7 +80,7 @@ func (c *accrualCollector) mapBreakagePlansToFBOCollectionSources( expiresAt := plan.ExpiresAt route := plan.FBOAddress.Route().Route() for _, reservedSource := range reservedSources { - breakageSources = append(breakageSources, fboCollectionSource{ + breakageSources = append(breakageSources, source{ address: plan.FBOAddress, sourceChargeID: reservedSource.sourceChargeID, available: reservedSource.available, @@ -136,11 +108,11 @@ func (c *accrualCollector) mapBreakagePlansToFBOCollectionSources( // to one open breakage plan in memory. The ledger write happens later, when the // selected source is collected and its attached plan is released. func reserveSourcesForBreakagePlan( - sources []fboCollectionSource, + sources []source, plan breakage.Plan, currency currencies.CurrencyReference, targetRoute ledger.Route, -) []fboCollectionSource { +) []source { route := plan.FBOAddress.Route().Route() if !route.Currency.Equal(currency) { return nil @@ -157,30 +129,28 @@ func reserveSourcesForBreakagePlan( return reserveSourceUnknownBreakagePlan(sources, plan) } -func (c *accrualCollector) listCustomerFBOBalanceBucketSources( +func (c *service) listCustomerFBOBalanceBucketSources( ctx context.Context, - namespace string, + query SourceQuery, accountID string, - currency currencies.CurrencyReference, - targetRoute ledger.Route, - asOf time.Time, -) ([]fboCollectionSource, error) { +) ([]source, error) { // Query at source-charge granularity. Route/sub-account balance alone is too // coarse once multiple purchased credit sources share the same FBO route. route := ledger.RouteFilter{ - Currency: currency, + Currency: query.Currency, } + targetRoute := ledger.Route{Filters: query.Filters} if len(targetRoute.Filters.Features) == 1 { route.MatchFeature = targetRoute.Filters.Features[0] } else if len(targetRoute.Filters.Features) == 0 { route.Features = mo.Some([]string(nil)) } - buckets, err := c.deps.BalanceQuerier.GetBalanceBuckets(ctx, ledger.BalanceBucketQuery{ - Namespace: namespace, + buckets, err := c.GetSpendableBalanceBuckets(ctx, SpendableBalanceBucketQuery{ + Namespace: query.CustomerID.Namespace, Filters: ledger.Filters{ AccountID: &accountID, - AsOf: &asOf, + AsOf: &query.AsOf, Route: route, }, GroupBy: []string{ledger.BalanceBucketGroupBySourceChargeID}, @@ -189,7 +159,7 @@ func (c *accrualCollector) listCustomerFBOBalanceBucketSources( return nil, fmt.Errorf("get FBO balance buckets: %w", err) } - sources := make([]fboCollectionSource, 0, len(buckets)) + sources := make([]source, 0, len(buckets)) for _, bucket := range buckets { if !bucket.SettledAmount.IsPositive() { continue @@ -200,13 +170,13 @@ func (c *accrualCollector) listCustomerFBOBalanceBucketSources( continue } - source := fboCollectionSource{ + source := source{ address: bucket.Address, sourceChargeID: bucket.GroupByValues[ledger.BalanceBucketGroupBySourceChargeID], available: bucket.SettledAmount, creditPriority: customerFBOPriority(route), restricted: !route.Filters.IsEmpty(), - cursor: fboBalanceBucketCursor(bucket), + cursor: balanceBucketCursor(bucket), } sources = append(sources, source) } @@ -214,7 +184,7 @@ func (c *accrualCollector) listCustomerFBOBalanceBucketSources( return sources, nil } -func fboBalanceBucketCursor(bucket ledger.BalanceBucket) string { +func balanceBucketCursor(bucket ledger.BalanceBucket) string { sourceChargeID := lo.FromPtrOr(bucket.GroupByValues[ledger.BalanceBucketGroupBySourceChargeID], "null") return bucket.Address.SubAccountID() + ":" + sourceChargeID @@ -222,7 +192,7 @@ func fboBalanceBucketCursor(bucket ledger.BalanceBucket) string { // reserveSourceIdentifiedBreakagePlan maps to at most one live source bucket // because FBO balance buckets are grouped by sub-account + source_charge_id. -func reserveSourceIdentifiedBreakagePlan(sources []fboCollectionSource, plan breakage.Plan) []fboCollectionSource { +func reserveSourceIdentifiedBreakagePlan(sources []source, plan breakage.Plan) []source { if plan.SourceChargeID == nil { return nil } @@ -241,7 +211,7 @@ func reserveSourceIdentifiedBreakagePlan(sources []fboCollectionSource, plan bre return nil } - return []fboCollectionSource{reserved} + return []source{reserved} } return nil @@ -250,9 +220,9 @@ func reserveSourceIdentifiedBreakagePlan(sources []fboCollectionSource, plan bre // reserveSourceUnknownBreakagePlan handles source-less breakage records. Without // source_charge_id, the plan may reserve from multiple live source buckets in // its FBO sub-account, but the total reservation is capped by plan.OpenAmount. -func reserveSourceUnknownBreakagePlan(sources []fboCollectionSource, plan breakage.Plan) []fboCollectionSource { +func reserveSourceUnknownBreakagePlan(sources []source, plan breakage.Plan) []source { remaining := plan.OpenAmount - reservedSources := make([]fboCollectionSource, 0) + reservedSources := make([]source, 0) for i := range sources { if !remaining.IsPositive() { return reservedSources @@ -274,27 +244,27 @@ func reserveSourceUnknownBreakagePlan(sources []fboCollectionSource, plan breaka return reservedSources } -func reserveFBOBalanceBucketSource(source *fboCollectionSource, amount alpacadecimal.Decimal) (fboCollectionSource, bool) { - if !source.available.IsPositive() || !amount.IsPositive() { - return fboCollectionSource{}, false +func reserveFBOBalanceBucketSource(bucketSource *source, amount alpacadecimal.Decimal) (source, bool) { + if !bucketSource.available.IsPositive() || !amount.IsPositive() { + return source{}, false } - reserved := source.available + reserved := bucketSource.available if reserved.GreaterThan(amount) { reserved = amount } - source.available = source.available.Sub(reserved) + bucketSource.available = bucketSource.available.Sub(reserved) - out := *source + out := *bucketSource out.available = reserved return out, true } -func selectFBOSources(sources []fboCollectionSource, target alpacadecimal.Decimal) []fboCollectionSelection { +func selectFBOSources(sources []source, target alpacadecimal.Decimal) []selection { remaining := target - out := make([]fboCollectionSelection, 0, len(sources)) + out := make([]selection, 0, len(sources)) for _, source := range sources { if !remaining.IsPositive() { @@ -310,7 +280,7 @@ func selectFBOSources(sources []fboCollectionSource, target alpacadecimal.Decima amount = remaining } - out = append(out, fboCollectionSelection{ + out = append(out, selection{ source: source, amount: amount, }) diff --git a/openmeter/ledger/collector/types.go b/openmeter/ledger/fbo/sources_types.go similarity index 82% rename from openmeter/ledger/collector/types.go rename to openmeter/ledger/fbo/sources_types.go index 288927473a..5ece300a1e 100644 --- a/openmeter/ledger/collector/types.go +++ b/openmeter/ledger/fbo/sources_types.go @@ -1,4 +1,4 @@ -package collector +package fbo import ( "cmp" @@ -16,7 +16,7 @@ import ( // FBO Sources for Prioritization -type fboCollectionSource struct { +type source struct { address ledger.PostingAddress sourceChargeID *string available alpacadecimal.Decimal @@ -27,11 +27,11 @@ type fboCollectionSource struct { breakagePlan *breakage.Plan } -var _ cmpx.Comparable[fboCollectionSource] = fboCollectionSource{} +var _ cmpx.Comparable[source] = source{} // TODO: Version this contract before changing it. Corrections and breakage // releases depend on collection selecting sources deterministically. -func (s fboCollectionSource) Compare(other fboCollectionSource) int { +func (s source) Compare(other source) int { if c := cmp.Compare(s.creditPriority, other.creditPriority); c != 0 { return c } @@ -64,15 +64,15 @@ func (s fboCollectionSource) Compare(other fboCollectionSource) int { // Selections for Consumption Plan -type fboCollectionSelection struct { +type selection struct { collectionOriginID *string - source fboCollectionSource + source source amount alpacadecimal.Decimal } -type fboCollectionSelections []fboCollectionSelection +type selectionList []selection -func (s fboCollectionSelections) postingAmounts(spendChargeID *string) []transactions.PostingAmount { +func (s selectionList) postingAmounts(spendChargeID *string) []transactions.PostingAmount { out := make([]transactions.PostingAmount, 0, len(s)) for idx, selection := range s { diff --git a/openmeter/ledger/historical/adapter/balancebuckets.go b/openmeter/ledger/historical/adapter/balancebuckets.go index 784da2d462..d218a829a9 100644 --- a/openmeter/ledger/historical/adapter/balancebuckets.go +++ b/openmeter/ledger/historical/adapter/balancebuckets.go @@ -45,33 +45,48 @@ func (r *repo) GetBalanceBuckets(ctx context.Context, query ledger.BalanceBucket return nil, err } - rows, err := tx.db.QueryContext(ctx, sqlQuery, args...) + return tx.queryBalanceBuckets(ctx, sqlQuery, args, query.GroupBy) + }) +} + +func (r *repo) GetMinimumBalanceBuckets(ctx context.Context, query ledger.MinimumBalanceBucketQuery) ([]ledger.BalanceBucket, error) { + return entutils.TransactingRepo(ctx, r, func(ctx context.Context, tx *repo) ([]ledger.BalanceBucket, error) { + sqlQuery, args, err := minimumBalanceBucketsQuery{query: query}.SQL() if err != nil { - return nil, fmt.Errorf("failed to query ledger balance buckets: %w", err) + return nil, err } - defer rows.Close() - buckets := make([]ledger.BalanceBucket, 0) - for rows.Next() { - row := balanceBucketRow{} - if err := rows.Scan(row.destinations()...); err != nil { - return nil, fmt.Errorf("failed to scan ledger balance bucket: %w", err) - } + return tx.queryBalanceBuckets(ctx, sqlQuery, args, query.GroupBy) + }) +} - bucket, err := row.toBalanceBucket(query.GroupBy) - if err != nil { - return nil, err - } +func (r *repo) queryBalanceBuckets(ctx context.Context, sqlQuery string, args []any, groupBy []string) ([]ledger.BalanceBucket, error) { + rows, err := r.db.QueryContext(ctx, sqlQuery, args...) + if err != nil { + return nil, fmt.Errorf("failed to query ledger balance buckets: %w", err) + } + defer rows.Close() - buckets = append(buckets, bucket) + buckets := make([]ledger.BalanceBucket, 0) + for rows.Next() { + row := balanceBucketRow{} + if err := rows.Scan(row.destinations()...); err != nil { + return nil, fmt.Errorf("failed to scan ledger balance bucket: %w", err) } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("ledger balance bucket rows: %w", err) + bucket, err := row.toBalanceBucket(groupBy) + if err != nil { + return nil, err } - return buckets, nil - }) + buckets = append(buckets, bucket) + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("ledger balance bucket rows: %w", err) + } + + return buckets, nil } func (r *balanceBucketRow) destinations() []any { diff --git a/openmeter/ledger/historical/adapter/balancebuckets_query.go b/openmeter/ledger/historical/adapter/balancebuckets_query.go index 79c1050fdd..3e4a22b295 100644 --- a/openmeter/ledger/historical/adapter/balancebuckets_query.go +++ b/openmeter/ledger/historical/adapter/balancebuckets_query.go @@ -37,6 +37,17 @@ func (q balanceBucketsQuery) SQL() (string, []any, error) { buckets := sql.With(balanceBucketCTEName).As(bucketSelector) buckets.SetDialect(dialect.Postgres) + query, args := selectBalanceBuckets().Prefix(buckets).Query() + + return query, args, nil +} + +// Load the account and route fields needed to turn each bucket into a posting +// address. +func selectBalanceBuckets() *sql.Selector { + buckets := sql.Table(balanceBucketCTEName) + buckets.SetDialect(dialect.Postgres) + subAccounts := sql.Table(ledgersubaccountdb.Table).As("sub_accounts") subAccounts.SetDialect(dialect.Postgres) @@ -73,7 +84,6 @@ func (q balanceBucketsQuery) SQL() (string, []any, error) { On(subAccounts.C(ledgersubaccountdb.FieldAccountID), accounts.C(ledgeraccountdb.FieldID)). Join(routes). On(subAccounts.C(ledgersubaccountdb.FieldRouteID), routes.C(ledgersubaccountroutedb.FieldID)). - Prefix(buckets). OrderBy( buckets.C(balanceBucketFieldSubAccountID), buckets.C(balanceBucketFieldSourceChargeID), @@ -82,9 +92,7 @@ func (q balanceBucketsQuery) SQL() (string, []any, error) { ) selector.SetDialect(dialect.Postgres) - sqlQuery, args := selector.Query() - - return sqlQuery, args, nil + return selector } func (q balanceBucketsQuery) bucketSelector() (*sql.Selector, error) { diff --git a/openmeter/ledger/historical/adapter/balancebuckets_test.go b/openmeter/ledger/historical/adapter/balancebuckets_test.go index 2bbd9dc0e3..4877729cc5 100644 --- a/openmeter/ledger/historical/adapter/balancebuckets_test.go +++ b/openmeter/ledger/historical/adapter/balancebuckets_test.go @@ -11,6 +11,7 @@ import ( "github.com/stretchr/testify/require" "github.com/openmeterio/openmeter/openmeter/currencies" + ledgerentrydb "github.com/openmeterio/openmeter/openmeter/ent/db/ledgerentry" "github.com/openmeterio/openmeter/openmeter/ledger" ledgeraccount "github.com/openmeterio/openmeter/openmeter/ledger/account" ledgerhistorical "github.com/openmeterio/openmeter/openmeter/ledger/historical" @@ -461,3 +462,85 @@ func chargeIDKeyPart(chargeID *string) string { return *chargeID } + +func TestRepo_GetMinimumBalanceBuckets_Timeline(t *testing.T) { + // All offsets are relative to the requested boundary. Rows are deliberately + // booked in the supplied order, independently of their effective time. + type movement struct { + offset time.Duration + amount int64 + annotations models.Annotations + } + + ignored := models.Annotations{"test.movement": "ignored"} + otherIgnored := models.Annotations{"test.other": "ignored"} + for _, tc := range []struct { + name string + movements []movement + expected float64 + }{ + {name: "future spending is reserved", movements: []movement{{-time.Hour, 10, nil}, {time.Hour, -7, nil}}, expected: 3}, + {name: "later return cannot hide earlier minimum", movements: []movement{{-time.Hour, 10, nil}, {time.Hour, -7, nil}, {2 * time.Hour, 7, nil}}, expected: 3}, + {name: "future only issuance starts at zero", movements: []movement{{time.Hour, 10, nil}}, expected: 0}, + {name: "future only deficit is retained", movements: []movement{{time.Hour, -4, nil}}, expected: -4}, + {name: "future replenishment cannot be spent early", movements: []movement{{-time.Hour, 4, nil}, {time.Hour, 6, nil}, {2 * time.Hour, -6, nil}}, expected: 4}, + {name: "simultaneous postings are netted", movements: []movement{{-time.Hour, 10, nil}, {time.Hour, -15, nil}, {time.Hour, 10, nil}}, expected: 5}, + {name: "boundary postings are inclusive", movements: []movement{{-time.Hour, 10, nil}, {0, -6, nil}, {time.Hour, 20, nil}}, expected: 4}, + {name: "out of order history uses effective time", movements: []movement{{2 * time.Hour, 7, nil}, {-time.Hour, 10, nil}, {time.Hour, -7, nil}}, expected: 3}, + {name: "excluded future debits and credits", movements: []movement{{-time.Hour, 10, nil}, {time.Hour, -10, ignored}, {2 * time.Hour, 6, ignored}, {3 * time.Hour, -4, nil}}, expected: 6}, + {name: "excluded future entry retains recorded age", movements: []movement{{time.Hour, -10, ignored}, {-time.Hour, 10, nil}}, expected: 10}, + {name: "future only excluded movement preserves the bucket", movements: []movement{{time.Hour, -10, ignored}}, expected: 0}, + {name: "either exclusion omits a future movement", movements: []movement{{-time.Hour, 10, nil}, {time.Hour, -4, ignored}, {2 * time.Hour, -3, otherIgnored}, {3 * time.Hour, -2, nil}}, expected: 8}, + {name: "different annotation value is included", movements: []movement{{-time.Hour, 10, nil}, {time.Hour, -4, models.Annotations{"test.movement": "included"}}}, expected: 6}, + {name: "exclusion does not affect opening balance", movements: []movement{{-2 * time.Hour, 10, nil}, {-time.Hour, -4, ignored}}, expected: 6}, + {name: "exclusion does not affect boundary", movements: []movement{{-time.Hour, 10, nil}, {0, -10, ignored}}, expected: 0}, + } { + t.Run(tc.name, func(t *testing.T) { + env := NewTestEnv(t) + t.Cleanup(func() { env.Close(t) }) + namespace := testNamespace() + at := time.Date(2026, time.January, 10, 0, 0, 0, 0, time.UTC) + route := ledger.Route{Currency: currencies.NewCurrencyReference(currencyx.Code("USD"))} + account := env.createSubAccountOfType(t, namespace, ledger.AccountTypeCustomerFBO, route) + counterpart := env.createSubAccountOfType(t, namespace, ledger.AccountTypeWash, route) + group, err := env.repo.CreateTransactionGroup(t.Context(), ledgerhistorical.CreateTransactionGroupInput{Namespace: namespace}) + require.NoError(t, err) + + // given: a persisted bucket with past and future movements. + for _, movement := range tc.movements { + amount := alpacadecimal.NewFromInt(movement.amount) + input := &transactionstestutils.AnyTransactionInput{ + BookedAtValue: at.Add(movement.offset), + AnnotationsValue: movement.annotations, + EntryInputsValues: []*transactionstestutils.AnyEntryInput{ + provenanceEntryInput(t, account, amount, nil, nil), + provenanceEntryInput(t, counterpart, amount.Neg(), nil, nil), + }, + } + _, err = env.repo.BookTransaction(t.Context(), models.NamespacedID{Namespace: namespace, ID: group.ID}, input) + require.NoError(t, err) + } + + // when: only later matching movements are excluded from the minimum. + buckets, err := env.repo.GetMinimumBalanceBuckets(t.Context(), ledger.MinimumBalanceBucketQuery{ + BalanceBucketQuery: ledger.BalanceBucketQuery{ + Namespace: namespace, + Filters: ledger.Filters{AccountID: &account.AccountID, AsOf: &at}, + GroupBy: []string{ledger.BalanceBucketGroupBySourceChargeID}, + }, + ExcludeFutureAnnotationFilters: map[string]string{"test.movement": "ignored", "test.other": "ignored"}, + }) + require.NoError(t, err) + + // then: the lowest effective balance includes the opening position. + require.Len(t, buckets, 1) + require.Equal(t, tc.expected, buckets[0].SettledAmount.InexactFloat64()) + require.Equal(t, tc.expected, buckets[0].PendingAmount.InexactFloat64()) + + oldest, err := env.client.LedgerEntry.Query().Where(ledgerentrydb.SubAccountID(account.ID)).Order(ledgerentrydb.ByCreatedAt()).First(t.Context()) + require.NoError(t, err) + require.True(t, oldest.CreatedAt.Equal(buckets[0].OldestMatchingEntryCreatedAt)) + require.Nil(t, buckets[0].GroupByValues[ledger.BalanceBucketGroupBySourceChargeID]) + }) + } +} diff --git a/openmeter/ledger/historical/adapter/minimumbalancebuckets_query.go b/openmeter/ledger/historical/adapter/minimumbalancebuckets_query.go new file mode 100644 index 0000000000..df7c5833b7 --- /dev/null +++ b/openmeter/ledger/historical/adapter/minimumbalancebuckets_query.go @@ -0,0 +1,178 @@ +package adapter + +import ( + "maps" + "slices" + + "entgo.io/ent/dialect" + "entgo.io/ent/dialect/sql" + "entgo.io/ent/dialect/sql/sqljson" + + ledgerentrydb "github.com/openmeterio/openmeter/openmeter/ent/db/ledgerentry" + ledgertransactiondb "github.com/openmeterio/openmeter/openmeter/ent/db/ledgertransaction" + "github.com/openmeterio/openmeter/openmeter/ledger" +) + +type minimumBalanceBucketsQuery struct { + query ledger.MinimumBalanceBucketQuery +} + +// Start with each bucket's balance at AsOf, add its later movements in time +// order, then take the lowest balance. Sum the past entries first so the running +// sum doesn't have to process the whole history. +// Future exclusions only affect later movements, not the starting balance. +func (q minimumBalanceBucketsQuery) SQL() (string, []any, error) { + if err := q.query.Validate(); err != nil { + return "", nil, err + } + + entryScope, err := q.scopedEntries() + if err != nil { + return "", nil, err + } + + movements := q.openingBalances(entryScope).UnionAll(q.futureMovements(entryScope)) + ctes := sql.Dialect(dialect.Postgres).With("movements").As(movements). + With("running_balances").As(q.runningBalances()). + With(balanceBucketCTEName).As(q.minimumBalances()) + query, args := selectBalanceBuckets().Prefix(ctes).Query() + + return query, args, nil +} + +// Apply the shared filters, but keep entries after AsOf: we need those to see +// how low the balance gets later. +func (q minimumBalanceBucketsQuery) scopedEntries() (*sql.Selector, error) { + filters := q.query.Filters + filters.AsOf = nil + predicates, err := (&sumEntriesQuery{query: ledger.Query{ + Namespace: q.query.Namespace, + Filters: filters, + }}).entryPredicates() + if err != nil { + return nil, err + } + + for _, key := range slices.Sorted(maps.Keys(q.query.ExcludeAnnotationFilters)) { + predicates = append(predicates, ledgerentrydb.HasTransactionWith(transactionAnnotationNotEqual(key, q.query.ExcludeAnnotationFilters[key]))) + } + + d := sql.Dialect(dialect.Postgres) + entries := d.Table(ledgerentrydb.Table) + txs := d.Table(ledgertransactiondb.Table).As("transactions") + scope := d.Select().From(entries). + Join(txs).On(entries.C(ledgerentrydb.FieldTransactionID), txs.C(ledgertransactiondb.FieldID)) + for _, predicate := range predicates { + predicate(scope) + } + + return scope, nil +} + +// Sum everything booked at or before AsOf. This is the starting balance for +// each bucket, so future-only exclusions don't apply here. +func (q minimumBalanceBucketsQuery) openingBalances(scopedEntries *sql.Selector) *sql.Selector { + d := sql.Dialect(dialect.Postgres) + entries := d.Table(ledgerentrydb.Table) + txs := d.Table(ledgertransactiondb.Table).As("transactions") + groupColumns := q.entryBucketColumns() + at := *q.query.Filters.AsOf + + return scopedEntries.Clone().Select(groupColumns...). + AppendSelectExprAs(sql.ExprFunc(func(b *sql.Builder) { + b.Arg(at).Join(sql.Raw("::timestamptz")) + }), "effective_at"). + AppendSelect( + sql.As(sql.Sum(entries.C(ledgerentrydb.FieldAmount)), "amount"), + sql.As(sql.Min(entries.C(ledgerentrydb.FieldCreatedAt)), balanceBucketFieldOldestMatchingEntryCreatedAt), + ). + Where(sql.LTE(txs.C(ledgertransactiondb.FieldBookedAt), at)).GroupBy(groupColumns...) +} + +// Combine entries with the same booked time before calculating balances. +// Otherwise, processing a debit before a credit could show a dip that never +// actually existed. Excluded entries count as zero so we still return their +// bucket and keep its oldest entry timestamp. +func (q minimumBalanceBucketsQuery) futureMovements(scopedEntries *sql.Selector) *sql.Selector { + d := sql.Dialect(dialect.Postgres) + entries := d.Table(ledgerentrydb.Table) + txs := d.Table(ledgertransactiondb.Table).As("transactions") + groupColumns := q.entryBucketColumns() + bookedAt := txs.C(ledgertransactiondb.FieldBookedAt) + amount := entries.C(ledgerentrydb.FieldAmount) + + futureAmount := sql.ExprFunc(func(b *sql.Builder) { + if len(q.query.ExcludeFutureAnnotationFilters) == 0 { + b.Ident(sql.Sum(amount)) + + return + } + + annotations := txs.C(ledgertransactiondb.FieldAnnotations) + included := make([]*sql.Predicate, 0, len(q.query.ExcludeFutureAnnotationFilters)) + for _, key := range slices.Sorted(maps.Keys(q.query.ExcludeFutureAnnotationFilters)) { + included = append(included, sql.Or( + sql.Not(sqljson.HasKey(annotations, sqljson.Path(key))), + sqljson.ValueNEQ(annotations, q.query.ExcludeFutureAnnotationFilters[key], sqljson.Path(key)), + )) + } + + b.Join(sql.Raw("SUM(CASE WHEN ")).Join(sql.And(included...)). + Join(sql.Raw(" THEN ")).Ident(amount).Join(sql.Raw(" ELSE 0 END)")) + }) + + return scopedEntries.Clone().Select(groupColumns...). + AppendSelect(sql.As(bookedAt, "effective_at")). + AppendSelectExprAs(futureAmount, "amount"). + AppendSelect(sql.As(sql.Min(entries.C(ledgerentrydb.FieldCreatedAt)), balanceBucketFieldOldestMatchingEntryCreatedAt)). + Where(sql.GT(bookedAt, *q.query.Filters.AsOf)).GroupBy(groupColumns...).GroupBy(bookedAt) +} + +// Add each later change to the starting balance, separately for each bucket. +// Entries at the same time have already been combined. +func (q minimumBalanceBucketsQuery) runningBalances() *sql.Selector { + d := sql.Dialect(dialect.Postgres) + groupColumns := q.bucketColumns() + + return d.Select(groupColumns...).AppendSelect("effective_at", balanceBucketFieldOldestMatchingEntryCreatedAt). + AppendSelectExprAs(sql.Window(func(b *sql.Builder) { + b.Ident(sql.Sum("amount")) + }).PartitionBy(groupColumns...).OrderBy("effective_at"), "balance").From(d.Table("movements")) +} + +// Take the lowest balance. If a bucket has no entries at AsOf, include zero: +// credit that only arrives later isn't available at AsOf. Negative balances +// still need to come through as negative. +func (q minimumBalanceBucketsQuery) minimumBalances() *sql.Selector { + d := sql.Dialect(dialect.Postgres) + groupColumns := q.bucketColumns() + + balances := d.Table("running_balances") + buckets := d.Select(balances.C(ledgerentrydb.FieldSubAccountID)).From(balances) + appendBalanceBucketDimensionSelect(buckets, balances, q.query.GroupBy, ledger.BalanceBucketGroupBySourceChargeID, ledgerentrydb.FieldSourceChargeID) + appendBalanceBucketDimensionSelect(buckets, balances, q.query.GroupBy, ledger.BalanceBucketGroupBySpendChargeID, ledgerentrydb.FieldSpendChargeID) + appendBalanceBucketDimensionSelect(buckets, balances, q.query.GroupBy, ledger.BalanceBucketGroupByCollectionOriginID, ledgerentrydb.FieldCollectionOriginID) + + buckets.AppendSelectExprAs(sql.ExprFunc(func(b *sql.Builder) { + b.Join(sql.Raw("CASE WHEN ")).Ident(sql.Min("effective_at")).WriteOp(sql.OpGT).Arg(*q.query.Filters.AsOf). + Join(sql.Raw(" THEN LEAST(MIN(balance), 0) ELSE MIN(balance) END")) + }), balanceBucketFieldSumAmount).AppendSelect( + sql.As(sql.Min(balanceBucketFieldOldestMatchingEntryCreatedAt), balanceBucketFieldOldestMatchingEntryCreatedAt), + ).GroupBy(groupColumns...) + + return buckets +} + +func (q minimumBalanceBucketsQuery) bucketColumns() []string { + return append([]string{ledgerentrydb.FieldSubAccountID}, q.query.GroupBy...) +} + +func (q minimumBalanceBucketsQuery) entryBucketColumns() []string { + entries := sql.Dialect(dialect.Postgres).Table(ledgerentrydb.Table) + columns := q.bucketColumns() + for idx, column := range columns { + columns[idx] = entries.C(column) + } + + return columns +} diff --git a/openmeter/ledger/historical/adapter/minimumbalancebuckets_query_test.go b/openmeter/ledger/historical/adapter/minimumbalancebuckets_query_test.go new file mode 100644 index 0000000000..5b61e5918d --- /dev/null +++ b/openmeter/ledger/historical/adapter/minimumbalancebuckets_query_test.go @@ -0,0 +1,291 @@ +package adapter + +import ( + "strings" + "testing" + "time" + + "github.com/samber/lo" + "github.com/stretchr/testify/require" + + "github.com/openmeterio/openmeter/openmeter/ledger" +) + +func TestMinimumBalanceBucketsQuery_SQLTemplate(t *testing.T) { + at := time.Date(2026, time.January, 10, 0, 0, 0, 0, time.UTC) + query, args, err := (minimumBalanceBucketsQuery{query: ledger.MinimumBalanceBucketQuery{ + BalanceBucketQuery: ledger.BalanceBucketQuery{ + Namespace: "ns-test", + Filters: ledger.Filters{AccountID: lo.ToPtr("account"), AsOf: &at}, + GroupBy: []string{ledger.BalanceBucketGroupBySourceChargeID}, + }, + ExcludeFutureAnnotationFilters: map[string]string{"movement": "ignored"}, + }}).SQL() + require.NoError(t, err) + expectedSQL := ` +WITH "movements" AS (SELECT + "ledger_entries"."sub_account_id", "ledger_entries"."source_charge_id", + ($1::timestamptz) AS "effective_at", + SUM("ledger_entries"."amount") AS "amount", + MIN("ledger_entries"."created_at") AS "oldest_matching_entry_created_at" + FROM "ledger_entries" + JOIN "ledger_transactions" AS "transactions" + ON "ledger_entries"."transaction_id" = "transactions"."id" + WHERE "ledger_entries"."namespace" = $2 AND EXISTS (SELECT "ledger_sub_accounts"."id" + FROM "ledger_sub_accounts" + WHERE "ledger_entries"."sub_account_id" = "ledger_sub_accounts"."id" + AND "ledger_sub_accounts"."account_id" = $3) AND "transactions"."booked_at" <= $4 + GROUP BY "ledger_entries"."sub_account_id", "ledger_entries"."source_charge_id" + UNION ALL + SELECT + "ledger_entries"."sub_account_id", "ledger_entries"."source_charge_id", + "transactions"."booked_at" AS "effective_at", + (SUM(CASE WHEN (NOT ("transactions"."annotations"->'movement' IS NOT NULL)) + OR "transactions"."annotations"->>'movement' <> $5 THEN "ledger_entries"."amount" ELSE 0 END)) AS "amount", + MIN("ledger_entries"."created_at") AS "oldest_matching_entry_created_at" + FROM "ledger_entries" + JOIN "ledger_transactions" AS "transactions" + ON "ledger_entries"."transaction_id" = "transactions"."id" + WHERE "ledger_entries"."namespace" = $6 AND EXISTS (SELECT "ledger_sub_accounts"."id" + FROM "ledger_sub_accounts" + WHERE "ledger_entries"."sub_account_id" = "ledger_sub_accounts"."id" + AND "ledger_sub_accounts"."account_id" = $7) AND "transactions"."booked_at" > $8 + GROUP BY "ledger_entries"."sub_account_id", "ledger_entries"."source_charge_id", "transactions"."booked_at"), +"running_balances" AS (SELECT + "sub_account_id", "source_charge_id", "effective_at", + "oldest_matching_entry_created_at", + (SUM("amount") OVER (PARTITION BY "sub_account_id", "source_charge_id" + ORDER BY "effective_at")) AS "balance" + FROM "movements"), +"balance_buckets" AS (SELECT + "running_balances"."sub_account_id", + "running_balances"."source_charge_id", + NULL AS "spend_charge_id", + NULL AS "collection_origin_id", + (CASE WHEN MIN("effective_at") > $9 THEN LEAST(MIN(balance), 0) ELSE MIN(balance) END) AS "sum_amount", + MIN("oldest_matching_entry_created_at") AS "oldest_matching_entry_created_at" + FROM "running_balances" + GROUP BY "sub_account_id", "source_charge_id") +SELECT + "balance_buckets"."sub_account_id", + "balance_buckets"."source_charge_id", + "balance_buckets"."spend_charge_id", + "balance_buckets"."collection_origin_id", + "balance_buckets"."sum_amount", + "balance_buckets"."oldest_matching_entry_created_at", + "sub_accounts"."route_id", + "accounts"."account_type", + "routes"."routing_key_version", + "routes"."routing_key", + "routes"."currency", + "routes"."cost_basis_currency", + "routes"."tax_code", + "routes"."tax_behavior", + "routes"."filters", + "routes"."cost_basis", + "routes"."credit_priority", + "routes"."transaction_authorization_status" +FROM "balance_buckets" +JOIN "ledger_sub_accounts" AS "sub_accounts" + ON "balance_buckets"."sub_account_id" = "sub_accounts"."id" +JOIN "ledger_accounts" AS "accounts" + ON "sub_accounts"."account_id" = "accounts"."id" +JOIN "ledger_sub_account_routes" AS "routes" + ON "sub_accounts"."route_id" = "routes"."id" +ORDER BY + "balance_buckets"."sub_account_id", + "balance_buckets"."source_charge_id", + "balance_buckets"."spend_charge_id", + "balance_buckets"."collection_origin_id" +` + require.Equal(t, strings.Join(strings.Fields(expectedSQL), " "), query) + require.Equal(t, []any{at, "ns-test", "account", at, "ignored", "ns-test", "account", at, at}, args) +} + +func TestMinimumBalanceBucketsQuery_ScopedEntriesSQLTemplate(t *testing.T) { + q := newMinimumBalanceBucketsTestQuery() + scope, err := q.scopedEntries() + require.NoError(t, err) + + query, args := scope.Query() + expectedSQL := ` +SELECT * FROM "ledger_entries" +JOIN "ledger_transactions" AS "transactions" + ON "ledger_entries"."transaction_id" = "transactions"."id" +WHERE "ledger_entries"."namespace" = $1 AND EXISTS (SELECT "ledger_sub_accounts"."id" + FROM "ledger_sub_accounts" + WHERE "ledger_entries"."sub_account_id" = "ledger_sub_accounts"."id" + AND "ledger_sub_accounts"."account_id" = $2) +` + require.Equal(t, strings.Join(strings.Fields(expectedSQL), " "), query) + require.Equal(t, []any{"ns-test", "account"}, args) +} + +func TestMinimumBalanceBucketsQuery_OpeningBalancesSQLTemplate(t *testing.T) { + q := newMinimumBalanceBucketsTestQuery() + scope, err := q.scopedEntries() + require.NoError(t, err) + + query, args := q.openingBalances(scope).Query() + expectedSQL := ` +SELECT + "ledger_entries"."sub_account_id", "ledger_entries"."source_charge_id", + ($1::timestamptz) AS "effective_at", + SUM("ledger_entries"."amount") AS "amount", + MIN("ledger_entries"."created_at") AS "oldest_matching_entry_created_at" +FROM "ledger_entries" +JOIN "ledger_transactions" AS "transactions" + ON "ledger_entries"."transaction_id" = "transactions"."id" +WHERE "ledger_entries"."namespace" = $2 AND EXISTS (SELECT "ledger_sub_accounts"."id" + FROM "ledger_sub_accounts" + WHERE "ledger_entries"."sub_account_id" = "ledger_sub_accounts"."id" + AND "ledger_sub_accounts"."account_id" = $3) AND "transactions"."booked_at" <= $4 +GROUP BY "ledger_entries"."sub_account_id", "ledger_entries"."source_charge_id" +` + require.Equal(t, strings.Join(strings.Fields(expectedSQL), " "), query) + require.Equal(t, []any{*q.query.Filters.AsOf, "ns-test", "account", *q.query.Filters.AsOf}, args) +} + +func TestMinimumBalanceBucketsQuery_FutureMovementsSQLTemplate(t *testing.T) { + t.Run("with future exclusions", func(t *testing.T) { + q := newMinimumBalanceBucketsTestQuery() + scope, err := q.scopedEntries() + require.NoError(t, err) + + query, args := q.futureMovements(scope).Query() + expectedSQL := ` +SELECT + "ledger_entries"."sub_account_id", "ledger_entries"."source_charge_id", + "transactions"."booked_at" AS "effective_at", + (SUM(CASE WHEN (NOT ("transactions"."annotations"->'movement' IS NOT NULL)) + OR "transactions"."annotations"->>'movement' <> $1 THEN "ledger_entries"."amount" ELSE 0 END)) AS "amount", + MIN("ledger_entries"."created_at") AS "oldest_matching_entry_created_at" +FROM "ledger_entries" +JOIN "ledger_transactions" AS "transactions" + ON "ledger_entries"."transaction_id" = "transactions"."id" +WHERE "ledger_entries"."namespace" = $2 AND EXISTS (SELECT "ledger_sub_accounts"."id" + FROM "ledger_sub_accounts" + WHERE "ledger_entries"."sub_account_id" = "ledger_sub_accounts"."id" + AND "ledger_sub_accounts"."account_id" = $3) AND "transactions"."booked_at" > $4 +GROUP BY "ledger_entries"."sub_account_id", "ledger_entries"."source_charge_id", "transactions"."booked_at" +` + require.Equal(t, strings.Join(strings.Fields(expectedSQL), " "), query) + require.Equal(t, []any{"ignored", "ns-test", "account", *q.query.Filters.AsOf}, args) + }) + + t.Run("without future exclusions", func(t *testing.T) { + q := newMinimumBalanceBucketsTestQuery() + q.query.ExcludeFutureAnnotationFilters = nil + scope, err := q.scopedEntries() + require.NoError(t, err) + + query, args := q.futureMovements(scope).Query() + expectedSQL := ` +SELECT + "ledger_entries"."sub_account_id", "ledger_entries"."source_charge_id", + "transactions"."booked_at" AS "effective_at", + (SUM("ledger_entries"."amount")) AS "amount", + MIN("ledger_entries"."created_at") AS "oldest_matching_entry_created_at" +FROM "ledger_entries" +JOIN "ledger_transactions" AS "transactions" + ON "ledger_entries"."transaction_id" = "transactions"."id" +WHERE "ledger_entries"."namespace" = $1 AND EXISTS (SELECT "ledger_sub_accounts"."id" + FROM "ledger_sub_accounts" + WHERE "ledger_entries"."sub_account_id" = "ledger_sub_accounts"."id" + AND "ledger_sub_accounts"."account_id" = $2) AND "transactions"."booked_at" > $3 +GROUP BY "ledger_entries"."sub_account_id", "ledger_entries"."source_charge_id", "transactions"."booked_at" +` + require.Equal(t, strings.Join(strings.Fields(expectedSQL), " "), query) + require.Equal(t, []any{"ns-test", "account", *q.query.Filters.AsOf}, args) + }) +} + +func TestMinimumBalanceBucketsQuery_RunningBalancesSQLTemplate(t *testing.T) { + q := newMinimumBalanceBucketsTestQuery() + query, args := q.runningBalances().Query() + expectedSQL := ` +SELECT + "sub_account_id", "source_charge_id", "effective_at", + "oldest_matching_entry_created_at", + (SUM("amount") OVER (PARTITION BY "sub_account_id", "source_charge_id" + ORDER BY "effective_at")) AS "balance" +FROM "movements" +` + require.Equal(t, strings.Join(strings.Fields(expectedSQL), " "), query) + require.Empty(t, args) +} + +func TestMinimumBalanceBucketsQuery_MinimumBalancesSQLTemplate(t *testing.T) { + q := newMinimumBalanceBucketsTestQuery() + query, args := q.minimumBalances().Query() + expectedSQL := ` +SELECT + "running_balances"."sub_account_id", + "running_balances"."source_charge_id", + NULL AS "spend_charge_id", + NULL AS "collection_origin_id", + (CASE WHEN MIN("effective_at") > $1 THEN LEAST(MIN(balance), 0) ELSE MIN(balance) END) AS "sum_amount", + MIN("oldest_matching_entry_created_at") AS "oldest_matching_entry_created_at" +FROM "running_balances" +GROUP BY "sub_account_id", "source_charge_id" +` + require.Equal(t, strings.Join(strings.Fields(expectedSQL), " "), query) + require.Equal(t, []any{*q.query.Filters.AsOf}, args) +} + +func TestMinimumBalanceBucketsQuery_BucketColumns(t *testing.T) { + for _, tc := range []struct { + name string + groupBy []string + columns []string + qualified []string + }{ + { + name: "subaccount only", + columns: []string{"sub_account_id"}, + qualified: []string{`"ledger_entries"."sub_account_id"`}, + }, + { + name: "source", + groupBy: []string{ledger.BalanceBucketGroupBySourceChargeID}, + columns: []string{"sub_account_id", "source_charge_id"}, + qualified: []string{`"ledger_entries"."sub_account_id"`, `"ledger_entries"."source_charge_id"`}, + }, + { + name: "all provenance dimensions", + groupBy: []string{ + ledger.BalanceBucketGroupBySourceChargeID, + ledger.BalanceBucketGroupBySpendChargeID, + ledger.BalanceBucketGroupByCollectionOriginID, + }, + columns: []string{"sub_account_id", "source_charge_id", "spend_charge_id", "collection_origin_id"}, + qualified: []string{ + `"ledger_entries"."sub_account_id"`, + `"ledger_entries"."source_charge_id"`, + `"ledger_entries"."spend_charge_id"`, + `"ledger_entries"."collection_origin_id"`, + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + q := newMinimumBalanceBucketsTestQuery() + q.query.GroupBy = tc.groupBy + + require.Equal(t, tc.columns, q.bucketColumns()) + require.Equal(t, tc.qualified, q.entryBucketColumns()) + }) + } +} + +func newMinimumBalanceBucketsTestQuery() minimumBalanceBucketsQuery { + at := time.Date(2026, time.January, 10, 0, 0, 0, 0, time.UTC) + + return minimumBalanceBucketsQuery{query: ledger.MinimumBalanceBucketQuery{ + BalanceBucketQuery: ledger.BalanceBucketQuery{ + Namespace: "ns-test", + Filters: ledger.Filters{AccountID: lo.ToPtr("account"), AsOf: &at}, + GroupBy: []string{ledger.BalanceBucketGroupBySourceChargeID}, + }, + ExcludeFutureAnnotationFilters: map[string]string{"movement": "ignored"}, + }} +} diff --git a/openmeter/ledger/historical/balance.go b/openmeter/ledger/historical/balance.go index dc79cf0a2a..dc9d7307cd 100644 --- a/openmeter/ledger/historical/balance.go +++ b/openmeter/ledger/historical/balance.go @@ -81,3 +81,11 @@ func (l *Ledger) GetBalancesAtBoundaries(ctx context.Context, input ledger.GetBa return l.repo.GetBalancesAtBoundaries(ctx, input) } + +func (l *Ledger) GetMinimumBalanceBuckets(ctx context.Context, query ledger.MinimumBalanceBucketQuery) ([]ledger.BalanceBucket, error) { + if err := query.Validate(); err != nil { + return nil, fmt.Errorf("validate minimum balance bucket query: %w", err) + } + + return l.repo.GetMinimumBalanceBuckets(ctx, query) +} diff --git a/openmeter/ledger/historical/repo.go b/openmeter/ledger/historical/repo.go index 12ec287cf7..e4e32ae114 100644 --- a/openmeter/ledger/historical/repo.go +++ b/openmeter/ledger/historical/repo.go @@ -33,6 +33,7 @@ type Repo interface { // Get balance buckets grouped by sub-account and requested dimensions. GetBalanceBuckets(ctx context.Context, query ledger.BalanceBucketQuery) ([]ledger.BalanceBucket, error) + GetMinimumBalanceBuckets(ctx context.Context, query ledger.MinimumBalanceBucketQuery) ([]ledger.BalanceBucket, error) // List transactions with pagination ListTransactions(ctx context.Context, input ledger.ListTransactionsInput) (ledger.ListTransactionsResult, error) diff --git a/openmeter/ledger/noop/noop.go b/openmeter/ledger/noop/noop.go index 64a8f20db2..71a553ce3a 100644 --- a/openmeter/ledger/noop/noop.go +++ b/openmeter/ledger/noop/noop.go @@ -315,3 +315,7 @@ func accountTypeForRoute(route ledger.Route) ledger.AccountType { return ledger.AccountTypeCustomerAccrued } } + +func (Ledger) GetMinimumBalanceBuckets(context.Context, ledger.MinimumBalanceBucketQuery) ([]ledger.BalanceBucket, error) { + return nil, nil +}