chore: move invoice pending lines to gathering types

openmeter/billing/adapter.go

@@ -81,13 +81,16 @@ type GatheringInvoiceAdapter interface {
 	DeleteGatheringInvoice(ctx context.Context, input DeleteGatheringInvoiceAdapterInput) error
 	GetGatheringInvoiceById(ctx context.Context, input GetGatheringInvoiceByIdInput) (GatheringInvoice, error)
 	ListGatheringInvoices(ctx context.Context, input ListGatheringInvoicesInput) (pagination.Result[GatheringInvoice], error)
+
+	HardDeleteGatheringInvoiceLines(ctx context.Context, invoiceID InvoiceID, lineIDs []string) error
 }
 
 type InvoiceSplitLineGroupAdapter interface {
 	CreateSplitLineGroup(ctx context.Context, input CreateSplitLineGroupAdapterInput) (SplitLineGroup, error)
 	UpdateSplitLineGroup(ctx context.Context, input UpdateSplitLineGroupInput) (SplitLineGroup, error)
 	DeleteSplitLineGroup(ctx context.Context, input DeleteSplitLineGroupInput) error
 	GetSplitLineGroup(ctx context.Context, input GetSplitLineGroupInput) (SplitLineHierarchy, error)
+	GetSplitLineGroupHeaders(ctx context.Context, input GetSplitLineGroupHeadersInput) (SplitLineGroupHeaders, error)
 }
 
 type SequenceAdapter interface {

openmeter/billing/adapter/gatheringinvoice.go

@@ -11,6 +11,7 @@ import (
 	"github.com/openmeterio/openmeter/openmeter/billing"
 	"github.com/openmeterio/openmeter/openmeter/ent/db"
 	"github.com/openmeterio/openmeter/openmeter/ent/db/billinginvoice"
+	"github.com/openmeterio/openmeter/openmeter/ent/db/billinginvoiceline"
 	"github.com/openmeterio/openmeter/pkg/clock"
 	"github.com/openmeterio/openmeter/pkg/convert"
 	"github.com/openmeterio/openmeter/pkg/framework/entutils"
@@ -121,7 +122,7 @@ func (a *adapter) UpdateGatheringInvoice(ctx context.Context, in billing.Gatheri
 			ClearPaymentProcessingEnteredAt().
 			ClearDraftUntil().
 			ClearIssuedAt().
-			ClearDeletedAt().
+			SetOrClearDeletedAt(convert.SafeToUTC(in.DeletedAt)).
 			ClearSentToCustomerAt().
 			ClearQuantitySnapshotedAt().
 			// Totals
@@ -139,9 +140,16 @@ func (a *adapter) UpdateGatheringInvoice(ctx context.Context, in billing.Gatheri
 			updateQuery = updateQuery.ClearCollectionAt()
 		}
 
-		updateQuery = updateQuery.
-			SetPeriodStart(in.ServicePeriod.From.In(time.UTC)).
-			SetPeriodEnd(in.ServicePeriod.To.In(time.UTC))
+		// Clear period when the invoice is soft-deleted
+		if in.DeletedAt != nil {
+			updateQuery = updateQuery.
+				ClearPeriodStart().
+				ClearPeriodEnd()
+		} else {
+			updateQuery = updateQuery.
+				SetPeriodStart(in.ServicePeriod.From.In(time.UTC)).
+				SetPeriodEnd(in.ServicePeriod.To.In(time.UTC))
+		}
 
 		// Supplier
 		updateQuery = updateQuery.
@@ -215,6 +223,9 @@ func (a *adapter) ListGatheringInvoices(ctx context.Context, input billing.ListG
 
 		if input.Expand.Has(billing.GatheringInvoiceExpandLines) {
 			query = query.WithBillingInvoiceLines(func(q *db.BillingInvoiceLineQuery) {
+				if !input.Expand.Has(billing.GatheringInvoiceExpandDeletedLines) {
+					q = q.Where(billinginvoiceline.DeletedAtIsNil())
+				}
 				q.WithUsageBasedLine()
 			})
 		}
@@ -337,6 +348,9 @@ func (a *adapter) GetGatheringInvoiceById(ctx context.Context, input billing.Get
 
 		if input.Expand.Has(billing.GatheringInvoiceExpandLines) {
 			query = query.WithBillingInvoiceLines(func(q *db.BillingInvoiceLineQuery) {
+				if !input.Expand.Has(billing.GatheringInvoiceExpandDeletedLines) {
+					q = q.Where(billinginvoiceline.DeletedAtIsNil())
+				}
 				q.WithUsageBasedLine()
 			})
 		}

openmeter/billing/adapter/gatheringlines.go

@@ -12,16 +12,94 @@ import (
 
 	"github.com/openmeterio/openmeter/openmeter/billing"
 	"github.com/openmeterio/openmeter/openmeter/ent/db"
+	"github.com/openmeterio/openmeter/openmeter/ent/db/billinginvoice"
 	"github.com/openmeterio/openmeter/openmeter/ent/db/billinginvoiceline"
 	"github.com/openmeterio/openmeter/openmeter/ent/db/billinginvoiceusagebasedlineconfig"
 	"github.com/openmeterio/openmeter/pkg/clock"
 	"github.com/openmeterio/openmeter/pkg/convert"
 	"github.com/openmeterio/openmeter/pkg/entitydiff"
+	"github.com/openmeterio/openmeter/pkg/framework/entutils"
 	"github.com/openmeterio/openmeter/pkg/models"
 	"github.com/openmeterio/openmeter/pkg/slicesx"
 	"github.com/openmeterio/openmeter/pkg/timeutil"
 )
 
+func (a *adapter) HardDeleteGatheringInvoiceLines(ctx context.Context, invoiceID billing.InvoiceID, lineIDs []string) error {
+	if err := invoiceID.Validate(); err != nil {
+		return fmt.Errorf("validating invoice ID: %w", err)
+	}
+
+	if len(lineIDs) == 0 {
+		return nil
+	}
+
+	return entutils.TransactingRepoWithNoValue(ctx, a, func(ctx context.Context, tx *adapter) error {
+		// Let's validate the delete
+		invoiceHeader, err := tx.db.BillingInvoice.Query().
+			Select(billinginvoice.FieldStatus, billinginvoice.FieldNamespace, billinginvoice.FieldCurrency).
+			Where(billinginvoice.ID(invoiceID.ID)).
+			Where(billinginvoice.Namespace(invoiceID.Namespace)).
+			Only(ctx)
+		if err != nil {
+			return err
+		}
+
+		if invoiceHeader.Status != billing.StandardInvoiceStatusGathering {
+			return fmt.Errorf("invoice is not a gathering invoice [id=%s, namespace=%s, currency=%s]", invoiceID.ID, invoiceID.Namespace, invoiceHeader.Currency)
+		}
+
+		// Let's determine the usage based line configs to delete
+		existingLines, err := tx.db.BillingInvoiceLine.Query().
+			Where(billinginvoiceline.InvoiceID(invoiceID.ID)).
+			Where(billinginvoiceline.Namespace(invoiceID.Namespace)).
+			Where(billinginvoiceline.IDIn(lineIDs...)).
+			WithUsageBasedLine().
+			All(ctx)
+		if err != nil {
+			return err
+		}
+
+		usageBasedLineConfigIDs, err := slicesx.MapWithErr(existingLines, func(line *db.BillingInvoiceLine) (string, error) {
+			if line.Edges.UsageBasedLine == nil {
+				return "", fmt.Errorf("usage based line is missing [line_id=%s]", line.ID)
+			}
+
+			return line.Edges.UsageBasedLine.ID, nil
+		})
+		if err != nil {
+			return err
+		}
+
+		nrDeleted, err := tx.db.BillingInvoiceLine.Delete().
+			Where(billinginvoiceline.InvoiceID(invoiceID.ID)).
+			Where(billinginvoiceline.Namespace(invoiceID.Namespace)).
+			Where(billinginvoiceline.IDIn(lineIDs...)).
+			Exec(ctx)
+		if err != nil {
+			return err
+		}
+
+		if nrDeleted != len(lineIDs) {
+			// Note: this causes a rollback of the transaction
+			return fmt.Errorf("failed to hard delete all gathering invoice lines [deleted=%d, linesToDelete=%d]", nrDeleted, len(lineIDs))
+		}
+
+		nrDeleted, err = tx.db.BillingInvoiceUsageBasedLineConfig.Delete().
+			Where(billinginvoiceusagebasedlineconfig.IDIn(usageBasedLineConfigIDs...)).
+			Where(billinginvoiceusagebasedlineconfig.Namespace(invoiceID.Namespace)).
+			Exec(ctx)
+		if err != nil {
+			return err
+		}
+
+		if nrDeleted != len(usageBasedLineConfigIDs) {
+			return fmt.Errorf("failed to hard delete all usage based line configs [deleted=%d, configsToDelete=%d]", nrDeleted, len(usageBasedLineConfigIDs))
+		}
+
+		return nil
+	})
+}
+
 type gatheringLineDiff struct {
 	Line entitydiff.Diff[*billing.GatheringLine]
 }

openmeter/billing/adapter/invoicelinesplitgroup.go

@@ -16,6 +16,8 @@ import (
 	"github.com/openmeterio/openmeter/pkg/timeutil"
 )
 
+var _ billing.InvoiceSplitLineGroupAdapter = (*adapter)(nil)
+
 func (a *adapter) CreateSplitLineGroup(ctx context.Context, input billing.CreateSplitLineGroupAdapterInput) (billing.SplitLineGroup, error) {
 	if err := input.Validate(); err != nil {
 		return billing.SplitLineGroup{}, billing.ValidationError{
@@ -295,3 +297,30 @@ func (a *adapter) fetchAllSplitLineGroups(ctx context.Context, namespace string,
 		return a.mapSplitLineHierarchyFromDB(ctx, dbSplitLineGroup)
 	})
 }
+
+func (a *adapter) GetSplitLineGroupHeaders(ctx context.Context, input billing.GetSplitLineGroupHeadersInput) (billing.SplitLineGroupHeaders, error) {
+	if err := input.Validate(); err != nil {
+		return billing.SplitLineGroupHeaders{}, billing.ValidationError{
+			Err: err,
+		}
+	}
+
+	return entutils.TransactingRepo(ctx, a, func(ctx context.Context, tx *adapter) (billing.SplitLineGroupHeaders, error) {
+		dbSplitLineGroups, err := tx.db.BillingInvoiceSplitLineGroup.Query().
+			Where(billinginvoicesplitlinegroup.Namespace(input.Namespace)).
+			Where(billinginvoicesplitlinegroup.IDIn(input.SplitLineGroupIDs...)).
+			All(ctx)
+		if err != nil {
+			return billing.SplitLineGroupHeaders{}, err
+		}
+
+		splitLineGroups, err := slicesx.MapWithErr(dbSplitLineGroups, func(dbSplitLineGroup *db.BillingInvoiceSplitLineGroup) (billing.SplitLineGroup, error) {
+			return a.mapSplitLineGroupFromDB(dbSplitLineGroup)
+		})
+		if err != nil {
+			return billing.SplitLineGroupHeaders{}, err
+		}
+
+		return splitLineGroups, nil
+	})
+}

openmeter/billing/gatheringinvoice.go

@@ -147,11 +147,13 @@ func (e GatheringInvoiceExpand) Validate() error {
 
 const (
 	GatheringInvoiceExpandLines            GatheringInvoiceExpand = "lines"
+	GatheringInvoiceExpandDeletedLines     GatheringInvoiceExpand = "deletedLines"
 	GatheringInvoiceExpandAvailableActions GatheringInvoiceExpand = "availableActions"
 )
 
 var GatheringInvoiceExpandValues = []GatheringInvoiceExpand{
 	GatheringInvoiceExpandLines,
+	GatheringInvoiceExpandDeletedLines,
 	GatheringInvoiceExpandAvailableActions,
 }
 
@@ -180,6 +182,18 @@ type GatheringInvoiceAvailableActions struct {
 
 type GatheringLines []GatheringLine
 
+func (l GatheringLines) Validate() error {
+	return errors.Join(
+		lo.Map(l, func(l GatheringLine, _ int) error {
+			err := l.Validate()
+			if err != nil {
+				return fmt.Errorf("line[%s]: %w", l.ID, err)
+			}
+			return nil
+		})...,
+	)
+}
+
 type GatheringInvoiceLines struct {
 	mo.Option[GatheringLines]
 }
@@ -189,15 +203,7 @@ func (l GatheringInvoiceLines) Validate() error {
 		return nil
 	}
 
-	return errors.Join(
-		lo.Map(l.OrEmpty(), func(l GatheringLine, _ int) error {
-			err := l.Validate()
-			if err != nil {
-				return fmt.Errorf("line[%s]: %w", l.ID, err)
-			}
-			return nil
-		})...,
-	)
+	return l.OrEmpty().Validate()
 }
 
 func (l *GatheringInvoiceLines) Sort() {
@@ -246,6 +252,54 @@ func (l *GatheringInvoiceLines) Append(lines ...GatheringLine) {
 	l.Option = mo.Some(append(l.OrEmpty(), lines...))
 }
 
+func (l GatheringInvoiceLines) GetReferencedFeatureKeys() ([]string, error) {
+	if l.IsAbsent() {
+		return nil, nil
+	}
+
+	keys := make([]string, 0, len(l.OrEmpty()))
+	for _, line := range l.OrEmpty() {
+		if line.FeatureKey == "" {
+			continue
+		}
+
+		keys = append(keys, line.FeatureKey)
+	}
+
+	return lo.Uniq(keys), nil
+}
+
+func (l GatheringInvoiceLines) GetByID(id string) (GatheringLine, bool) {
+	if l.IsAbsent() {
+		return GatheringLine{}, false
+	}
+
+	lines := l.OrEmpty()
+	for _, line := range lines {
+		if line.ID == id {
+			return line, true
+		}
+	}
+
+	return GatheringLine{}, false
+}
+
+func (l *GatheringInvoiceLines) SetByID(line GatheringLine) error {
+	if l.IsAbsent() {
+		return fmt.Errorf("lines are absent")
+	}
+
+	lines := l.OrEmpty()
+	for i := range lines {
+		if lines[i].ID == line.ID {
+			lines[i] = line
+			return nil
+		}
+	}
+
+	return fmt.Errorf("line[%s]: line not found", line.ID)
+}
+
 func NewGatheringInvoiceLines(children []GatheringLine) GatheringInvoiceLines {
 	return GatheringInvoiceLines{
 		Option: mo.Some(GatheringLines(children)),
@@ -371,6 +425,30 @@ func (i GatheringLineBase) Clone() (GatheringLineBase, error) {
 	return out, nil
 }
 
+func (i GatheringLineBase) GetFeatureKey() string {
+	return i.FeatureKey
+}
+
+func (i GatheringLineBase) GetServicePeriod() timeutil.ClosedPeriod {
+	return i.ServicePeriod
+}
+
+func (i GatheringLineBase) GetPrice() *productcatalog.Price {
+	return &i.Price
+}
+
+func (i GatheringLineBase) GetID() string {
+	return i.ID
+}
+
+func (i GatheringLineBase) GetInvoiceAt() time.Time {
+	return i.InvoiceAt
+}
+
+func (i GatheringLineBase) GetSplitLineGroupID() *string {
+	return i.SplitLineGroupID
+}
+
 // TODO: rename to GatheringLine
 type GatheringLine struct {
 	GatheringLineBase `json:",inline"`
@@ -390,6 +468,26 @@ func (g GatheringLine) Clone() (GatheringLine, error) {
 	}, nil
 }
 
+func (i GatheringLine) CloneForCreate(edits ...func(*GatheringLine)) (GatheringLine, error) {
+	clone, err := i.Clone()
+	if err != nil {
+		return GatheringLine{}, fmt.Errorf("cloning line: %w", err)
+	}
+
+	clone.ID = ""
+	clone.UBPConfigID = ""
+	clone.CreatedAt = time.Time{}
+	clone.UpdatedAt = time.Time{}
+	clone.DeletedAt = nil
+	clone.DBState = nil
+
+	for _, edit := range edits {
+		edit(&clone)
+	}
+
+	return clone, nil
+}
+
 func (g GatheringLine) WithoutDBState() (GatheringLine, error) {
 	clone, err := g.Clone()
 	if err != nil {
@@ -516,8 +614,10 @@ type ListGatheringInvoicesInput struct {
 func (i ListGatheringInvoicesInput) Validate() error {
 	var errs []error
 
-	if err := i.Page.Validate(); err != nil {
-		errs = append(errs, fmt.Errorf("page: %w", err))
+	if !lo.IsEmpty(i.Page) {
+		if err := i.Page.Validate(); err != nil {
+			errs = append(errs, fmt.Errorf("page: %w", err))
+		}
 	}
 
 	if len(i.Namespaces) == 0 {