package service import ( "context" "fmt" "time" "github.com/samber/lo" "github.com/samber/mo" "github.com/openmeterio/openmeter/openmeter/billing" "github.com/openmeterio/openmeter/openmeter/billing/charges/invoiceupdater" "github.com/openmeterio/openmeter/openmeter/billing/charges/meta" "github.com/openmeterio/openmeter/openmeter/billing/charges/usagebased" usagebasedrun "github.com/openmeterio/openmeter/openmeter/billing/charges/usagebased/service/run" "github.com/openmeterio/openmeter/openmeter/productcatalog" "github.com/openmeterio/openmeter/openmeter/streaming" "github.com/openmeterio/openmeter/pkg/clock" "github.com/openmeterio/openmeter/pkg/currencyx" "github.com/openmeterio/openmeter/pkg/models" "github.com/openmeterio/openmeter/pkg/ref" "github.com/openmeterio/openmeter/pkg/slicesx" ) var _ billing.LineEngine = (*LineEngine)(nil) type LineEngine struct { service *service } func (e *LineEngine) GetLineEngineType() billing.LineEngineType { return billing.LineEngineTypeChargeUsageBased } func (e *LineEngine) IsLineBillableAsOf(_ context.Context, input billing.IsLineBillableAsOfInput) (bool, error) { if err := input.Validate(); err != nil { return false, fmt.Errorf("validating input: %w", err) } return !input.AsOf.Before(input.ResolvedBillablePeriod.To), nil } func (e *LineEngine) SplitGatheringLine(_ context.Context, input billing.SplitGatheringLineInput) (billing.SplitGatheringLineResult, error) { res := billing.SplitGatheringLineResult{} if err := input.Validate(); err != nil { return res, fmt.Errorf("validating input: %w", err) } line := input.Line if line.ChargeID == nil || *line.ChargeID == "" { return res, fmt.Errorf("usage based gathering line[%s]: charge id is required", line.ID) } if !line.ServicePeriod.Contains(input.SplitAt) { return res, fmt.Errorf("usage based gathering line[%s]: splitAt is not within the line period", line.ID) } postSplitAtLine, err := line.CloneForCreate(func(l *billing.GatheringLine) { l.ServicePeriod.From = input.SplitAt l.ChildUniqueReferenceID = nil }) if err != nil { return res, fmt.Errorf("cloning post split line: %w", err) } postSplitAtLineEmpty, err := isUsageBasedSplitPeriodEmpty(postSplitAtLine) if err != nil { return res, fmt.Errorf("checking if post split line is empty: %w", err) } if !postSplitAtLineEmpty { if err := postSplitAtLine.Validate(); err != nil { return res, fmt.Errorf("validating post split line: %w", err) } } line.ServicePeriod.To = input.SplitAt line.InvoiceAt = input.SplitAt line.ChildUniqueReferenceID = nil preSplitAtLine := line preSplitAtLineEmpty, err := isUsageBasedSplitPeriodEmpty(preSplitAtLine) if err != nil { return res, fmt.Errorf("checking if pre split line is empty: %w", err) } if preSplitAtLineEmpty { preSplitAtLine.DeletedAt = lo.ToPtr(clock.Now()) } else { if err := preSplitAtLine.Validate(); err != nil { return res, fmt.Errorf("validating pre split line: %w", err) } } var postSplitAtLinePtr *billing.GatheringLine if !postSplitAtLineEmpty { postSplitAtLinePtr = &postSplitAtLine } return billing.SplitGatheringLineResult{ PreSplitAtLine: preSplitAtLine, PostSplitAtLine: postSplitAtLinePtr, }, nil } func (e *LineEngine) BuildStandardInvoiceLines(ctx context.Context, input billing.BuildStandardInvoiceLinesInput) (billing.StandardLines, error) { stdLines, err := slicesx.MapWithErr(input.GatheringLines, func(gatheringLine billing.GatheringLine) (*billing.StandardLine, error) { stdLine, err := gatheringLine.AsNewStandardLine(input.Invoice.ID) if err != nil { return nil, fmt.Errorf("converting gathering line to standard line: %w", err) } return stdLine, nil }) if err != nil { return nil, err } return stdLines, nil } func (e *LineEngine) BuildStandardLinesForGatheringPreview(ctx context.Context, input billing.BuildStandardInvoiceLinesInput) (billing.StandardLines, error) { if err := input.Validate(); err != nil { return nil, fmt.Errorf("validating input: %w", err) } stdLines, err := input.GatheringLines.ToStandardLines(input.Invoice.ID) if err != nil { return nil, err } chargesByID, err := e.getChargesForStandardLineEvent(ctx, billing.StandardLineEventInput{ Invoice: input.Invoice, Lines: stdLines, }, meta.Expands{ meta.ExpandRealizations, meta.ExpandDetailedLines, }, "gathering preview") if err != nil { return nil, err } for _, stdLine := range stdLines { charge, ok := chargesByID[*stdLine.ChargeID] if !ok { return nil, fmt.Errorf("usage based charge[%s] not found for gathering preview line[%s]", *stdLine.ChargeID, stdLine.ID) } previewResult, err := e.buildGatheringPreviewRun(ctx, charge, stdLine) if err != nil { return nil, fmt.Errorf("building gathering preview run for line[%s]: %w", stdLine.ID, err) } if err := populateStandardLineFromRun(stdLine, populateStandardLineFromRunInput{ Run: previewResult.Run, Runs: previewResult.Runs, }); err != nil { return nil, fmt.Errorf("populating gathering preview line[%s] from run: %w", stdLine.ID, err) } if err := stdLine.Validate(); err != nil { return nil, fmt.Errorf("validating gathering preview line[%s]: %w", stdLine.ID, err) } } return stdLines, nil } func (e *LineEngine) buildGatheringPreviewRun(ctx context.Context, charge usagebased.Charge, stdLine *billing.StandardLine) (usagebasedrun.BuildCreditThenInvoiceGatheringPreviewRunResult, error) { if charge.Intent.GetSettlementMode() != productcatalog.CreditThenInvoiceSettlementMode { return usagebasedrun.BuildCreditThenInvoiceGatheringPreviewRunResult{}, fmt.Errorf( "usage based standard line[%s]: unsupported settlement mode for gathering preview: %s", stdLine.ID, charge.Intent.GetSettlementMode(), ) } stateMachineConfig, err := e.service.getStateMachineConfigForCharge(ctx, charge) if err != nil { return usagebasedrun.BuildCreditThenInvoiceGatheringPreviewRunResult{}, fmt.Errorf("getting state machine config for line[%s]: %w", stdLine.ID, err) } runType := getInvoiceRealizationRunType(charge, stdLine.Period) storedAtLT := meta.NormalizeTimestamp(stdLine.Period.To) servicePeriodTo := storedAtLT if runType == usagebased.RealizationRunTypeFinalRealization { storedAtLT, _ = stateMachineConfig.CustomerOverride.MergedProfile.WorkflowConfig.Collection.Interval.AddTo(charge.Intent.GetEffectiveServicePeriod().To) storedAtLT = meta.NormalizeTimestamp(storedAtLT) servicePeriodTo = meta.NormalizeTimestamp(charge.Intent.GetEffectiveServicePeriod().To) } return e.service.runs.BuildCreditThenInvoiceGatheringPreviewRun(ctx, usagebasedrun.BuildCreditThenInvoiceGatheringPreviewRunInput{ Charge: charge, CustomerOverride: stateMachineConfig.CustomerOverride, FeatureMeter: stateMachineConfig.FeatureMeter, Type: runType, StoredAtLT: storedAtLT, ServicePeriodTo: servicePeriodTo, LineID: stdLine.ID, InvoiceID: stdLine.InvoiceID, CurrencyCalculator: stateMachineConfig.CurrencyCalculator, }) } func (e *LineEngine) OnStandardInvoiceCreated(ctx context.Context, input billing.OnStandardInvoiceCreatedInput) (billing.StandardLines, error) { if err := input.Validate(); err != nil { return nil, fmt.Errorf("validating input: %w", err) } stdLines, err := slicesx.MapWithErr(input.Lines, func(stdLine *billing.StandardLine) (*billing.StandardLine, error) { stateMachine, err := e.newStateMachineForStandardLine(ctx, stdLine) if err != nil { return nil, err } if stateMachine.GetCharge().Intent.GetSettlementMode() != productcatalog.CreditThenInvoiceSettlementMode { return nil, fmt.Errorf( "usage based standard line[%s]: unsupported settlement mode for standard invoice creation: %s", stdLine.ID, stateMachine.GetCharge().Intent.GetSettlementMode(), ) } // Becoming active after the service period starts is not an invoice lifecycle event, so we // still rely on the generic TriggerNext/AdvanceUntilStateStable flow before invoice-created // lifecycle transitions take over. if _, err := stateMachine.AdvanceUntilStateStable(ctx); err != nil { return nil, fmt.Errorf("advancing usage based charge[%s]: %w", stateMachine.GetCharge().ID, err) } if stateMachine.GetCharge().State.CurrentRealizationRunID != nil { return nil, billing.ValidationError{ Err: fmt.Errorf("line[%s]: %w", stdLine.ID, usagebased.ErrActiveRealizationRunAlreadyExists), } } if err := stateMachine.FireAndActivate(ctx, meta.TriggerInvoiceCreated, invoiceCreatedInput{ LineID: stdLine.ID, InvoiceID: input.Invoice.ID, ServicePeriod: stdLine.Period, }); err != nil { return nil, fmt.Errorf("triggering %s for charge[%s]: %w", meta.TriggerInvoiceCreated, stateMachine.GetCharge().ID, err) } if _, err := stateMachine.AdvanceUntilStateStable(ctx); err != nil { return nil, fmt.Errorf("advancing usage based charge[%s] after %s: %w", stateMachine.GetCharge().ID, meta.TriggerInvoiceCreated, err) } charge := stateMachine.GetCharge() currentRun, err := charge.GetCurrentRealizationRun() if err != nil { return nil, fmt.Errorf("getting current realization run for charge[%s]: %w", charge.ID, err) } if err := populateStandardLineFromRun(stdLine, populateStandardLineFromRunInput{ Run: currentRun, Runs: charge.Realizations, }); err != nil { return nil, fmt.Errorf("populating standard line from run for charge[%s]: %w", charge.ID, err) } if err := stdLine.Validate(); err != nil { return nil, fmt.Errorf("validating standard line[%s]: %w", stdLine.ID, err) } return stdLine, nil }) if err != nil { return nil, err } return stdLines, nil } func (e *LineEngine) OnCollectionCompleted(ctx context.Context, input billing.OnCollectionCompletedInput) (billing.StandardLines, error) { if err := input.Validate(); err != nil { return nil, fmt.Errorf("validating input: %w", err) } for _, stdLine := range input.Lines { stateMachine, err := e.newStateMachineForStandardLine(ctx, stdLine) if err != nil { return nil, err } canFire, err := stateMachine.CanFire(ctx, meta.TriggerCollectionCompleted) if err != nil { return nil, fmt.Errorf("checking collection_completed for charge[%s]: %w", stateMachine.GetCharge().ID, err) } if !canFire { continue } if err := stateMachine.FireAndActivate(ctx, meta.TriggerCollectionCompleted); err != nil { return nil, fmt.Errorf("triggering collection_completed for charge[%s]: %w", stateMachine.GetCharge().ID, err) } if _, err := stateMachine.AdvanceUntilStateStable(ctx); err != nil { return nil, fmt.Errorf("advancing usage based charge[%s] after collection_completed: %w", stateMachine.GetCharge().ID, err) } charge := stateMachine.GetCharge() currentRun, err := charge.GetCurrentRealizationRun() if err != nil { return nil, fmt.Errorf("getting current realization run for charge[%s]: %w", charge.ID, err) } if err := populateStandardLineFromRun(stdLine, populateStandardLineFromRunInput{ Run: currentRun, Runs: charge.Realizations, }); err != nil { return nil, fmt.Errorf("populating standard line from run for charge[%s]: %w", charge.ID, err) } if err := stdLine.Validate(); err != nil { return nil, fmt.Errorf("validating standard line[%s]: %w", stdLine.ID, err) } } return input.Lines, nil } func (e *LineEngine) OnMutableInvoiceLinesEditedViaAPI(ctx context.Context, input billing.OnMutableInvoiceUpdateInput) (billing.OnMutableInvoiceUpdateResult, error) { if err := input.Validate(); err != nil { return billing.OnMutableInvoiceUpdateResult{}, fmt.Errorf("validating input: %w", err) } createdLines, err := e.createManualInvoiceLines(ctx, input) if err != nil { return billing.OnMutableInvoiceUpdateResult{}, err } if len(input.Updated) > 0 { return billing.OnMutableInvoiceUpdateResult{}, fmt.Errorf("usage-based charge update: %w", billing.ErrCannotUpdateChargeManagedLine) } for _, line := range input.Deleted { if err := e.handleInvoiceLineDeleteViaAPI(ctx, input.Invoice, line); err != nil { return billing.OnMutableInvoiceUpdateResult{}, err } } return billing.OnMutableInvoiceUpdateResult{ CreatedLines: createdLines, }, nil } func (e *LineEngine) ValidateMutableInvoiceLineEditViaAPI(ctx context.Context, input billing.OnMutableInvoiceUpdateInput) error { if err := input.Validate(); err != nil { return fmt.Errorf("validating input: %w", err) } for _, line := range input.Created { if _, err := intentFromManualCreatedLine(ctx, input.Invoice, line, input.DefaultTaxCodeResolvers.Invoicing); err != nil { if line == nil { return fmt.Errorf("building manually created usage-based charge intent: %w", err) } return fmt.Errorf("building manually created usage-based charge intent for line[%s]: %w", line.GetID(), err) } } if len(input.Updated) > 0 { return fmt.Errorf("usage-based charge update: %w", billing.ErrCannotUpdateChargeManagedLine) } for _, line := range input.Deleted { if _, err := e.validateInvoiceLineDeleteViaAPI(ctx, input.Invoice, line); err != nil { return err } } return nil } type manualCreatedInvoiceLine struct { sourceLine billing.GenericInvoiceLine intent usagebased.Intent } func (e *LineEngine) createManualInvoiceLines(ctx context.Context, input billing.OnMutableInvoiceUpdateInput) ([]billing.GenericInvoiceLine, error) { if len(input.Created) == 0 { return nil, nil } if input.Invoice == nil { return nil, fmt.Errorf("invoice is required") } created, err := lo.MapErr(input.Created, func(line billing.GenericInvoiceLine, _ int) (manualCreatedInvoiceLine, error) { intent, err := intentFromManualCreatedLine(ctx, input.Invoice, line, input.DefaultTaxCodeResolvers.Invoicing) if err != nil { if line == nil { return manualCreatedInvoiceLine{}, fmt.Errorf("building manually created usage-based charge intent: %w", err) } return manualCreatedInvoiceLine{}, fmt.Errorf("building manually created usage-based charge intent for line[%s]: %w", line.GetID(), err) } return manualCreatedInvoiceLine{ sourceLine: line, intent: intent, }, nil }) if err != nil { return nil, err } namespace := input.Invoice.GetInvoiceID().Namespace intents := lo.Map(created, func(line manualCreatedInvoiceLine, _ int) usagebased.Intent { return line.intent }) featureMeters, err := e.service.featureService.ResolveFeatureMeters(ctx, namespace, lo.Map(intents, func(intent usagebased.Intent, _ int) ref.IDOrKey { return ref.IDOrKey{Key: intent.FeatureKey} })...) if err != nil { return nil, fmt.Errorf("resolving manually created usage-based charge feature meters: %w", err) } createdCharges, err := e.service.Create(ctx, usagebased.CreateInput{ Namespace: namespace, Intents: intents, FeatureMeters: featureMeters, }) if err != nil { return nil, fmt.Errorf("creating manually managed usage-based charges: %w", err) } if len(createdCharges) != len(created) { return nil, fmt.Errorf("expected %d manually created usage-based charges, got %d", len(created), len(createdCharges)) } out, err := lo.MapErr(createdCharges, func(charge usagebased.ChargeWithGatheringLine, idx int) (billing.GenericInvoiceLine, error) { sourceLine := created[idx].sourceLine switch sourceLine.AsInvoiceLine().Type() { case billing.InvoiceLineTypeGathering: if charge.GatheringLineToCreate == nil { return nil, fmt.Errorf("line[%s]: manually created usage-based charge[%s] did not create a gathering line", sourceLine.GetID(), charge.Charge.ID) } line, err := sourceLine.WithTargetState(charge.GatheringLineToCreate.AsGenericLine()) if err != nil { return nil, fmt.Errorf("line[%s]: merging manually created usage-based charge target state: %w", sourceLine.GetID(), err) } return line, nil case billing.InvoiceLineTypeStandard: standardInvoice, err := input.Invoice.AsInvoice().AsStandardInvoice() if err != nil { return nil, fmt.Errorf("getting standard invoice for created line[%s]: %w", sourceLine.GetID(), err) } standardLine, err := sourceLine.AsInvoiceLine().AsStandardLine() if err != nil { return nil, fmt.Errorf("getting created standard line[%s]: %w", sourceLine.GetID(), err) } line, err := e.attachManualStandardLine(ctx, standardInvoice, standardLine, sourceLine, charge.Charge) if err != nil { return nil, err } return line, nil default: return nil, fmt.Errorf("unsupported manually created usage-based line type [charge_id=%s,line_id=%s,line_type=%s]: %w", charge.Charge.ID, sourceLine.GetID(), sourceLine.AsInvoiceLine().Type(), billing.ErrCannotUpdateChargeManagedLine) } }) if err != nil { return nil, err } return out, nil } func (e *LineEngine) attachManualStandardLine(ctx context.Context, standardInvoice billing.StandardInvoice, standardLine billing.StandardLine, sourceLine billing.GenericInvoiceLine, charge usagebased.Charge) (billing.GenericInvoiceLine, error) { stateMachine, err := e.service.newStateMachineForCharge(ctx, charge) if err != nil { return nil, fmt.Errorf("new state machine for usage-based charge[%s]: %w", charge.ID, err) } if _, err := stateMachine.AdvanceUntilStateStable(ctx); err != nil { return nil, fmt.Errorf("advancing usage-based charge[%s]: %w", charge.ID, err) } if stateMachine.GetCharge().State.CurrentRealizationRunID != nil { return nil, billing.ValidationError{ Err: fmt.Errorf("line[%s]: %w", sourceLine.GetID(), usagebased.ErrActiveRealizationRunAlreadyExists), } } if err := stateMachine.FireAndActivate(ctx, meta.TriggerInvoiceCreated, invoiceCreatedInput{ LineID: standardLine.ID, InvoiceID: standardInvoice.ID, ServicePeriod: standardLine.Period, }); err != nil { return nil, fmt.Errorf("triggering %s for charge[%s]: %w", meta.TriggerInvoiceCreated, charge.ID, err) } if patches := stateMachine.DrainInvoicePatches(); len(patches) > 0 { return nil, fmt.Errorf("line[%s]: expected no invoice patches while attaching manually created usage-based charge[%s], got %v", sourceLine.GetID(), charge.ID, patches) } charge = stateMachine.GetCharge() currentRun, err := charge.GetCurrentRealizationRun() if err != nil { return nil, fmt.Errorf("getting current realization run for charge[%s]: %w", charge.ID, err) } standardLine.ChargeID = lo.ToPtr(charge.ID) standardLine.Engine = billing.LineEngineTypeChargeUsageBased standardLine.ManagedBy = billing.ManuallyManagedLine if err := populateStandardLineFromRun(&standardLine, populateStandardLineFromRunInput{ Run: currentRun, Runs: charge.Realizations, }); err != nil { return nil, fmt.Errorf("populating standard line from run for charge[%s]: %w", charge.ID, err) } if err := standardLine.Validate(); err != nil { return nil, fmt.Errorf("validating standard line[%s]: %w", standardLine.ID, err) } line, err := sourceLine.WithTargetState(standardLine.AsGenericLine()) if err != nil { return nil, fmt.Errorf("line[%s]: merging manually created usage-based standard line target state: %w", sourceLine.GetID(), err) } return line, nil } func (e *LineEngine) validateInvoiceLineDeleteViaAPI(ctx context.Context, invoice billing.GenericInvoiceReader, line billing.GenericInvoiceLine) (usagebased.Charge, error) { if invoice == nil { return usagebased.Charge{}, fmt.Errorf("invoice is required") } chargeID := line.GetChargeID() if chargeID == nil || *chargeID == "" { return usagebased.Charge{}, fmt.Errorf("usage based line[%s]: charge id is required", line.GetID()) } charge, err := e.service.GetByID(ctx, usagebased.GetByIDInput{ ChargeID: meta.ChargeID{ Namespace: line.GetLineID().Namespace, ID: *chargeID, }, Expands: meta.Expands{ meta.ExpandRealizations, meta.ExpandDetailedLines, }, }) if err != nil { return usagebased.Charge{}, fmt.Errorf("getting usage based charge for deleted line[%s]: %w", line.GetID(), err) } if charge.Intent.GetSettlementMode() != productcatalog.CreditThenInvoiceSettlementMode { return usagebased.Charge{}, fmt.Errorf( "usage based line[%s]: unsupported settlement mode for API delete: %s", line.GetID(), charge.Intent.GetSettlementMode(), ) } nonVoidedRuns := charge.Realizations.WithoutVoidedBillingHistory() switch line.AsInvoiceLine().Type() { case billing.InvoiceLineTypeGathering: // No pre-validation is required, deletion is supported regardless of the charge state. case billing.InvoiceLineTypeStandard: if len(nonVoidedRuns) > 1 { return usagebased.Charge{}, fmt.Errorf("usage based standard line[%s] cannot be deleted with multiple realization runs: %w", line.GetID(), billing.ErrCannotEditProgressivelyBilledUsageBasedLine) } if len(nonVoidedRuns) == 0 { // This is an internal consistency error, we are not supposed to surface this to the user, so no typed error wrapping. return usagebased.Charge{}, fmt.Errorf("usage based standard line[%s] cannot be deleted with no realization runs", line.GetID()) } default: return usagebased.Charge{}, fmt.Errorf("usage based line[%s]: unexpected line type: %s", line.GetID(), line.AsInvoiceLine().Type()) } return charge, nil } func (e *LineEngine) handleInvoiceLineDeleteViaAPI(ctx context.Context, invoice billing.GenericInvoiceReader, line billing.GenericInvoiceLine) error { chargeID := line.GetChargeID() if chargeID == nil || *chargeID == "" { return fmt.Errorf("usage based line[%s]: charge id is required", line.GetID()) } charge, err := e.validateInvoiceLineDeleteViaAPI(ctx, invoice, line) if err != nil { return err } switch line.AsInvoiceLine().Type() { case billing.InvoiceLineTypeGathering: nonVoidedRuns := charge.Realizations.WithoutVoidedBillingHistory() var patch meta.Patch if len(nonVoidedRuns) > 0 { lineServicePeriod := line.GetServicePeriod() shrinkToRealizedPeriodPatch, err := meta.NewPatchShrinkToRealizedPeriod(meta.NewPatchShrinkToRealizedPeriodInput{ ChangeSource: billing.ChangeSourceAPIRequest, NewServicePeriodEnd: lineServicePeriod.From, }) if err != nil { return fmt.Errorf("creating usage based charge[%s] API shrink to realized period patch: %w", charge.ID, err) } patch = shrinkToRealizedPeriodPatch } else { deletePatch, err := meta.NewPatchDelete(meta.NewPatchDeleteInput{ ChangeSource: billing.ChangeSourceAPIRequest, Policy: meta.RefundAsCreditsDeletePolicy, }) if err != nil { return fmt.Errorf("creating usage based charge[%s] API delete patch: %w", charge.ID, err) } patch = deletePatch } _, patches, err := e.applyChargePatchForInvoiceLineEditViaAPI(ctx, charge, patch) if err != nil { return fmt.Errorf("usage based line[%s]: applying %s patch for charge[%s]: %w", line.GetID(), patch.Op(), charge.ID, err) } // The edited gathering invoice already deletes this line. The charge // state machine must still agree by emitting the same pending-line // deletion, which proves the API edit persisted the matching charge // intent change instead of leaving charge state behind. gatheringPatch, err := patches.RequireSingularGatheringLinePatchForCharge(*chargeID) if err != nil { return fmt.Errorf("line[%s]: validating gathering-line API delete patch target: %w", line.GetID(), err) } if gatheringPatch.Op() != invoiceupdater.PatchOpDeleteGatheringLineByChargeID { return fmt.Errorf("line[%s]: expected gathering-line delete patch, got %s", line.GetID(), gatheringPatch.Op()) } return nil case billing.InvoiceLineTypeStandard: deletePatch, err := meta.NewPatchDelete(meta.NewPatchDeleteInput{ ChangeSource: billing.ChangeSourceAPIRequest, Policy: meta.RefundAsCreditsDeletePolicy, }) if err != nil { return fmt.Errorf("creating usage based charge[%s] API delete patch: %w", charge.ID, err) } charge, patches, err := e.applyChargePatchForInvoiceLineEditViaAPI(ctx, charge, deletePatch) if err != nil { return fmt.Errorf("usage based line[%s]: applying charge delete patch for charge[%s]: %w", line.GetID(), charge.ID, err) } standardInvoice, err := invoice.AsInvoice().AsStandardInvoice() if err != nil { return fmt.Errorf("usage based line[%s]: getting standard invoice: %w", line.GetID(), err) } stdInvoicePatches, rest, err := patches.BisectByStandardInvoiceID(standardInvoice.ID) if err != nil { return fmt.Errorf("usage based line[%s]: bisecting invoice patches for charge[%s]: %w", line.GetID(), charge.ID, err) } if len(stdInvoicePatches) != 1 { return fmt.Errorf("received unexpected number of standard invoice patches for line[%s]: count=%d %v", line.GetID(), len(stdInvoicePatches), stdInvoicePatches) } stdInvoicePatch, err := stdInvoicePatches.RequireSingularStandardInvoiceLineDeletePatch() if err != nil { return fmt.Errorf("usage based line[%s]: requiring singular standard invoice line delete patch for charge[%s]: %w", line.GetID(), charge.ID, err) } if err := stdInvoicePatch.RequireTarget(line); err != nil { return fmt.Errorf("usage based line[%s]: validating standard invoice line delete patch target for charge[%s]: %w", line.GetID(), charge.ID, err) } standardLine, err := line.AsInvoiceLine().AsStandardLine() if err != nil { return fmt.Errorf("usage based line[%s]: getting standard line for charge[%s]: %w", line.GetID(), charge.ID, err) } _, err = e.deleteMutableStandardLineRealization(ctx, charge, standardInvoice, &standardLine) if err != nil { return fmt.Errorf("usage based line[%s]: deleting mutable standard line realization for charge[%s]: %w", line.GetID(), charge.ID, err) } // Handle the remaining gathering line patches if err := rest.RequireType(invoiceupdater.PatchOpDeleteGatheringLineByChargeID, invoiceupdater.CountLessThanOrEqualTo(1)); err != nil { return fmt.Errorf("usage based line[%s]: validating remaining gathering line delete patches for charge[%s]: %w", line.GetID(), charge.ID, err) } if len(rest) > 0 { err := e.service.invoiceUpdater.ApplyPatches(ctx, invoice.GetCustomerID(), rest) if err != nil { return fmt.Errorf("usage based line[%s]: applying remaining gathering line delete patches for charge[%s]: %w", line.GetID(), charge.ID, err) } } return nil default: return fmt.Errorf("usage based line[%s]: unexpected line type: %s", line.GetID(), line.AsInvoiceLine().Type()) } } func (e *LineEngine) applyChargePatchForInvoiceLineEditViaAPI(ctx context.Context, charge usagebased.Charge, patch meta.Patch) (usagebased.Charge, invoiceupdater.Patches, error) { if err := patch.Validate(); err != nil { return usagebased.Charge{}, nil, fmt.Errorf("validating usage based charge[%s] API line edit patch: %w", charge.ID, err) } stateMachine, err := e.service.newStateMachineForCharge(ctx, charge) if err != nil { return usagebased.Charge{}, nil, fmt.Errorf("new state machine for usage based charge[%s]: %w", charge.ID, err) } if err := stateMachine.FireAndActivate(ctx, patch.Trigger(), patch); err != nil { return usagebased.Charge{}, nil, fmt.Errorf("triggering %s for charge[%s]: %w", patch.Trigger(), charge.ID, err) } return stateMachine.GetCharge(), stateMachine.DrainInvoicePatches(), nil } func (e *LineEngine) OnMutableStandardLinesDeletedBySystem(ctx context.Context, input billing.OnMutableStandardLinesDeletedInput) error { if err := input.Validate(); err != nil { return fmt.Errorf("validating input: %w", err) } chargesByID, err := e.getChargesForStandardLineEvent(ctx, input, meta.Expands{ meta.ExpandRealizations, meta.ExpandDetailedLines, }, "deleted standard lines") if err != nil { return err } // Whole-invoice deletion needs to remove the leftover gathering line for the // same charge. Charge patch updates can delete a mutable standard line while // also emitting replacement gathering-line patches, so this hook must not // apply an extra delete for ordinary system line updates. isInvoiceDelete := input.Invoice.DeletionSource != "" gatheringLineDeletePatches := make(invoiceupdater.Patches, 0, len(input.Lines)) for _, stdLine := range input.Lines { charge, ok := chargesByID[*stdLine.ChargeID] if !ok { return fmt.Errorf("usage based charge[%s] not found for deleted standard line[%s]", *stdLine.ChargeID, stdLine.ID) } charge, err = e.deleteMutableStandardLineRealization(ctx, charge, input.Invoice, stdLine) if err != nil { return err } chargesByID[*stdLine.ChargeID] = charge if isInvoiceDelete { gatheringLineDeletePatches = append(gatheringLineDeletePatches, invoiceupdater.NewDeleteGatheringLineByChargeIDPatch(*stdLine.ChargeID)) } } if len(gatheringLineDeletePatches) > 0 { if err := e.service.invoiceUpdater.ApplyPatches(ctx, input.Invoice.GetCustomerID(), gatheringLineDeletePatches); err != nil { return fmt.Errorf("applying gathering line delete patches for deleted usage based standard lines: %w", err) } } return nil } // deleteMutableStandardLineRealization removes the usage-based realization // backing a mutable deleted standard invoice line, including credit correction // and current-run detachment. func (e *LineEngine) deleteMutableStandardLineRealization( ctx context.Context, charge usagebased.Charge, invoice billing.StandardInvoice, stdLine *billing.StandardLine, ) (usagebased.Charge, error) { run, err := charge.Realizations.GetByLineID(stdLine.ID) if err != nil { return usagebased.Charge{}, err } // Deleted realizations have already been cleaned up through a prior line deletion, // so billing must not run the cleanup path for them again. if run.DeletedAt != nil { return usagebased.Charge{}, fmt.Errorf("usage based standard line[%s] cannot be deleted because realization run[%s] is already deleted", stdLine.ID, run.ID.ID) } if run.InvoiceID == nil || *run.InvoiceID != invoice.ID { return usagebased.Charge{}, fmt.Errorf("usage based standard line[%s] cannot be deleted because realization run[%s] is not associated with invoice[%s]", stdLine.ID, run.ID.ID, invoice.ID) } if run.Payment != nil { return usagebased.Charge{}, fmt.Errorf("usage based standard line[%s] cannot be deleted because realization run[%s] has payment allocation", stdLine.ID, run.ID.ID) } if run.InvoiceUsage != nil { return usagebased.Charge{}, fmt.Errorf("usage based standard line[%s] cannot be deleted because realization run[%s] has invoice accrued allocation", stdLine.ID, run.ID.ID) } cur, err := currencyx.NewCurrencyBuilder(currencyx.CurrencyTypeFiat). WithCode(charge.Intent.GetCurrency()). Build() if err != nil { return usagebased.Charge{}, fmt.Errorf("getting currency calculator for charge[%s]: %w", charge.ID, err) } now := clock.Now() if _, err := e.service.runs.CorrectAllCredits(ctx, usagebasedrun.CorrectAllCreditRealizationsInput{ Charge: charge, Run: run, AllocateAt: run.ServicePeriodTo, CurrencyCalculator: cur, }); err != nil { return usagebased.Charge{}, fmt.Errorf("correcting credits for deleted usage based standard line[%s] run[%s]: %w", stdLine.ID, run.ID.ID, err) } charge, err = e.markMutableStandardLineRunDeleted(ctx, charge, run, now) if err != nil { return usagebased.Charge{}, fmt.Errorf("marking realization run[%s] deleted for usage based standard line[%s]: %w", run.ID.ID, stdLine.ID, err) } return charge, nil } func (e *LineEngine) OnUnsupportedCreditNote(ctx context.Context, input billing.OnUnsupportedCreditNoteInput) error { if err := input.Validate(); err != nil { return fmt.Errorf("validating input: %w", err) } chargesByID, err := e.getChargesForStandardLineEvent(ctx, input, meta.Expands{ meta.ExpandRealizations, }, "unsupported credit note") if err != nil { return err } for _, stdLine := range input.Lines { charge, ok := chargesByID[*stdLine.ChargeID] if !ok { return fmt.Errorf("usage based charge[%s] not found for unsupported credit note line[%s]", *stdLine.ChargeID, stdLine.ID) } // Unsupported credit notes void the run for future rating history, but // they must not mark it deleted; deleted runs mean invoice/ledger cleanup // already happened, while this state preserves audit history. run, err := charge.Realizations.GetByLineID(stdLine.ID) if err != nil { return err } if run.InvoiceID == nil || *run.InvoiceID != input.Invoice.ID { return fmt.Errorf("usage based standard line[%s] cannot be marked unsupported credit note because realization run[%s] is not associated with invoice[%s]", stdLine.ID, run.ID.ID, input.Invoice.ID) } if run.DeletedAt != nil { return fmt.Errorf("usage based standard line[%s] cannot be marked unsupported credit note because realization run[%s] is already deleted", stdLine.ID, run.ID.ID) } if run.Type == usagebased.RealizationRunTypeInvalidDueToUnsupportedCreditNote { continue } // We need to mark the run as invalid to prevent it from being considered in further realization runs. if _, err := e.service.adapter.UpdateRealizationRun(ctx, usagebased.UpdateRealizationRunInput{ ID: run.ID, Type: mo.Some(usagebased.RealizationRunTypeInvalidDueToUnsupportedCreditNote), }); err != nil { return fmt.Errorf("marking realization run[%s] invalid due to unsupported credit note for usage based standard line[%s]: %w", run.ID.ID, stdLine.ID, err) } } return nil } func (e *LineEngine) markMutableStandardLineRunDeleted( ctx context.Context, charge usagebased.Charge, run usagebased.RealizationRun, deletedAt time.Time, ) (usagebased.Charge, error) { if _, err := e.service.adapter.UpdateRealizationRun(ctx, usagebased.UpdateRealizationRunInput{ ID: run.ID, DeletedAt: mo.Some(lo.ToPtr(deletedAt)), }); err != nil { return usagebased.Charge{}, err } charge.Realizations = charge.Realizations.Without(run.ID) currentRunDeleted := charge.State.CurrentRealizationRunID != nil && *charge.State.CurrentRealizationRunID == run.ID.ID if currentRunDeleted { charge.State.CurrentRealizationRunID = nil if charge.Status != usagebased.StatusDeleted { charge.Status = usagebased.StatusActive charge.State.AdvanceAfter = lo.ToPtr(meta.NormalizeTimestamp(charge.Intent.GetEffectiveServicePeriod().To)) } updatedChargeBase, err := e.service.adapter.UpdateCharge(ctx, charge.ChargeBase) if err != nil { return usagebased.Charge{}, err } charge.ChargeBase = updatedChargeBase } return charge, nil } func (e *LineEngine) getChargesForStandardLineEvent(ctx context.Context, input billing.StandardLineEventInput, expands meta.Expands, operation string) (map[string]usagebased.Charge, error) { chargeIDs := make([]string, 0, len(input.Lines)) seenChargeIDs := make(map[string]struct{}, len(input.Lines)) for _, stdLine := range input.Lines { if stdLine.ChargeID == nil || *stdLine.ChargeID == "" { return nil, fmt.Errorf("usage based standard line[%s]: charge id is required", stdLine.ID) } if stdLine.Namespace != input.Invoice.Namespace { return nil, fmt.Errorf("usage based standard line[%s]: namespace %s does not match invoice namespace %s", stdLine.ID, stdLine.Namespace, input.Invoice.Namespace) } if _, ok := seenChargeIDs[*stdLine.ChargeID]; ok { continue } seenChargeIDs[*stdLine.ChargeID] = struct{}{} chargeIDs = append(chargeIDs, *stdLine.ChargeID) } charges, err := e.service.GetByIDs(ctx, usagebased.GetByIDsInput{ Namespace: input.Invoice.Namespace, IDs: chargeIDs, Expands: expands, }) if err != nil { return nil, fmt.Errorf("getting usage based charges for %s: %w", operation, err) } return lo.KeyBy(charges, func(charge usagebased.Charge) string { return charge.ID }), nil } func (e *LineEngine) OnInvoiceIssued(ctx context.Context, input billing.OnInvoiceIssuedInput) error { if err := input.Validate(); err != nil { return fmt.Errorf("validating input: %w", err) } return e.fireLineTrigger(ctx, fireLineTriggerInput{ Lines: input.Lines, Trigger: meta.TriggerInvoiceIssued, InputFn: func(stdLine *billing.StandardLine) models.Validator { return billing.StandardLineWithInvoiceHeader{ Line: stdLine, Invoice: input.Invoice, } }, AdvanceUntilStateStable: true, }) } func (e *LineEngine) OnPaymentAuthorized(ctx context.Context, input billing.OnPaymentAuthorizedInput) error { if err := input.Validate(); err != nil { return fmt.Errorf("validating input: %w", err) } return e.recordRunPayments(ctx, recordRunPaymentsInput{ Lines: input.Lines, Invoice: input.Invoice, RecordFn: e.recordPaymentAuthorized, }) } func (e *LineEngine) OnPaymentSettled(ctx context.Context, input billing.OnPaymentSettledInput) error { if err := input.Validate(); err != nil { return fmt.Errorf("validating input: %w", err) } return e.recordRunPayments(ctx, recordRunPaymentsInput{ Lines: input.Lines, Invoice: input.Invoice, RecordFn: e.recordPaymentSettled, }) } type fireLineTriggerInput struct { Lines billing.StandardLines Trigger meta.Trigger InputFn func(*billing.StandardLine) models.Validator AdvanceUntilStateStable bool } func (i fireLineTriggerInput) Validate() error { if len(i.Lines) == 0 { return fmt.Errorf("lines are required") } if i.Trigger == "" { return fmt.Errorf("trigger is required") } if i.InputFn == nil { return fmt.Errorf("inputFn is required") } return nil } func (e *LineEngine) fireLineTrigger(ctx context.Context, input fireLineTriggerInput) error { if err := input.Validate(); err != nil { return fmt.Errorf("validating fire line trigger input: %w", err) } for _, stdLine := range input.Lines { stateMachine, err := e.newStateMachineForStandardLine(ctx, stdLine) if err != nil { return err } canFire, err := stateMachine.CanFire(ctx, input.Trigger) if err != nil { return fmt.Errorf("checking %s for charge[%s]: %w", input.Trigger, stateMachine.GetCharge().ID, err) } if !canFire { return fmt.Errorf( "charge[%s] in status %s cannot handle %s for standard line[%s]", stateMachine.GetCharge().ID, stateMachine.GetCharge().Status, input.Trigger, stdLine.ID, ) } if err := stateMachine.FireAndActivate(ctx, input.Trigger, input.InputFn(stdLine)); err != nil { return fmt.Errorf("triggering %s for charge[%s]: %w", input.Trigger, stateMachine.GetCharge().ID, err) } if input.AdvanceUntilStateStable { if _, err := stateMachine.AdvanceUntilStateStable(ctx); err != nil { return fmt.Errorf("advancing usage based charge[%s] after %s: %w", stateMachine.GetCharge().ID, input.Trigger, err) } } } return nil } func (e *LineEngine) newStateMachineForStandardLine(ctx context.Context, stdLine *billing.StandardLine) (StateMachine, error) { if stdLine.ChargeID == nil { return nil, fmt.Errorf("usage based standard line[%s]: charge id is required", stdLine.ID) } charge, err := e.service.GetByID(ctx, usagebased.GetByIDInput{ ChargeID: meta.ChargeID{ Namespace: stdLine.Namespace, ID: *stdLine.ChargeID, }, Expands: meta.Expands{meta.ExpandRealizations}, }) if err != nil { return nil, fmt.Errorf("getting usage based charge for line[%s]: %w", stdLine.ID, err) } stateMachine, err := e.service.newStateMachineForCharge(ctx, charge) if err != nil { return nil, fmt.Errorf("creating state machine for line[%s]: %w", stdLine.ID, err) } return stateMachine, nil } func isUsageBasedSplitPeriodEmpty(line billing.GatheringLine) (bool, error) { price := line.GetPrice() if price == nil { return false, fmt.Errorf("price is nil") } if price.Type() == productcatalog.FlatPriceType { return false, nil } return line.GetServicePeriod().Truncate(streaming.MinimumWindowSizeDuration).IsEmpty(), nil }