feat(billing): add collectionAt to invoices (#2175)

openmeter/billing/adapter/invoice.go

@@ -209,6 +209,13 @@ func (a *adapter) ListInvoices(ctx context.Context, input billing.ListInvoicesIn
 			query = query.Where(billinginvoice.DraftUntilLTE(*input.DraftUntil))
 		}
 
+		if input.CollectionAt != nil {
+			query = query.Where(billinginvoice.Or(
+				billinginvoice.CollectionAtLTE(*input.CollectionAt),
+				billinginvoice.CollectionAtIsNil(),
+			))
+		}
+
 		order := entutils.GetOrdering(sortx.OrderDefault)
 		if !input.Order.IsDefaultValue() {
 			order = entutils.GetOrdering(input.Order)
@@ -324,6 +331,11 @@ func (a *adapter) CreateInvoice(ctx context.Context, input billing.CreateInvoice
 			SetSupplierName(supplier.Name).
 			SetNillableSupplierTaxCode(supplier.TaxCode)
 
+		// Set collection_at only for new gathering invoices
+		if input.Status == billing.InvoiceStatusGathering {
+			createMut = createMut.SetCollectionAt(clock.Now())
+		}
+
 		if customer.BillingAddress != nil {
 			createMut = createMut.
 				// Customer contacts
@@ -455,6 +467,11 @@ func (a *adapter) UpdateInvoice(ctx context.Context, in billing.UpdateInvoiceAda
 			SetTaxesInclusiveTotal(in.Totals.TaxesInclusiveTotal).
 			SetTotal(in.Totals.Total)
 
+		if in.CollectionAt != nil && !in.CollectionAt.IsZero() {
+			updateQuery = updateQuery.
+				SetCollectionAt(*in.CollectionAt)
+		}
+
 		if in.Period != nil {
 			updateQuery = updateQuery.
 				SetPeriodStart(in.Period.Start).
@@ -680,6 +697,8 @@ func (a *adapter) mapInvoiceFromDB(ctx context.Context, invoice *db.BillingInvoi
 			UpdatedAt: invoice.UpdatedAt.In(time.UTC),
 			DeletedAt: convert.TimePtrIn(invoice.DeletedAt, time.UTC),
 
+			CollectionAt: lo.ToPtr(invoice.CollectionAt.In(time.UTC)),
+
 			ExternalIDs: billing.InvoiceExternalIDs{
 				Invoicing: lo.FromPtrOr(invoice.InvoicingAppExternalID, ""),
 				Payment:   lo.FromPtrOr(invoice.PaymentAppExternalID, ""),

openmeter/billing/invoice.go

@@ -205,6 +205,8 @@ type InvoiceBase struct {
 	IssuedAt   *time.Time `json:"issuedAt,omitempty"`
 	DeletedAt  *time.Time `json:"deletedAt,omitempty"`
 
+	CollectionAt *time.Time `json:"collectionAt,omitempty"`
+
 	// Customer is either a snapshot of the contact information of the customer at the time of invoice being sent
 	// or the data from the customer entity (draft state)
 	// This is required so that we are not modifying the invoice after it has been sent to the customer.
@@ -563,6 +565,10 @@ type ListInvoicesInput struct {
 	// Invoice is expired if the time defined by Invoice.DraftUntil is in the past compared to ListInvoicesInput.DraftUntil.
 	DraftUntil *time.Time
 
+	// CollectionAt allows to filter invoices which have their collection_at attribute is in the past compared
+	// to the time provided in CollectionAt parameter.
+	CollectionAt *time.Time
+
 	Expand InvoiceExpand
 
 	OrderBy api.InvoiceOrderBy

openmeter/billing/service/collectionat.go

@@ -0,0 +1,62 @@
+package billingservice
+
+import (
+	"time"
+
+	"github.com/samber/lo"
+
+	"github.com/openmeterio/openmeter/openmeter/billing"
+)
+
+// UpdateInvoiceCollectionAt updates the collectionAt attribute of the invoice with gathering type
+// using the customers collection configuration. It returns true if the attribute has been updated.
+// The collectionAt is calculated by adding the collection interval (from CollectionConfig) to the earliest invoicedAt
+// timestamp of the invoice lines on the gathering invoice.
+func UpdateInvoiceCollectionAt(invoice *billing.Invoice, collection billing.CollectionConfig) bool {
+	if invoice == nil || invoice.Status != billing.InvoiceStatusGathering {
+		return false
+	}
+
+	var invoiceAt time.Time
+
+	// Find the invoice lint with the earliest invoiceAt attribute
+	invoice.Lines.ForEach(func(v []*billing.Line) {
+		for _, line := range v {
+			if line == nil || line.Status != billing.InvoiceLineStatusValid {
+				continue
+			}
+
+			if line.DeletedAt != nil {
+				continue
+			}
+
+			if invoiceAt.IsZero() {
+				invoiceAt = line.InvoiceAt
+				continue
+			}
+
+			if line.InvoiceAt.Before(invoiceAt) {
+				invoiceAt = line.InvoiceAt
+			}
+		}
+	})
+
+	if invoiceAt.IsZero() {
+		return false
+	}
+
+	interval, ok := collection.Interval.Duration()
+	if !ok {
+		return false
+	}
+
+	collectionAt := invoiceAt.Add(interval)
+
+	if lo.FromPtr(invoice.CollectionAt).Equal(collectionAt) {
+		return false
+	}
+
+	invoice.CollectionAt = &collectionAt
+
+	return true
+}

openmeter/billing/service/invoice.go

@@ -14,7 +14,7 @@ import (
 	"github.com/samber/mo"
 
 	"github.com/openmeterio/openmeter/openmeter/billing"
-	lineservice "github.com/openmeterio/openmeter/openmeter/billing/service/lineservice"
+	"github.com/openmeterio/openmeter/openmeter/billing/service/lineservice"
 	customerentity "github.com/openmeterio/openmeter/openmeter/customer/entity"
 	"github.com/openmeterio/openmeter/pkg/clock"
 	"github.com/openmeterio/openmeter/pkg/currencyx"
@@ -791,6 +791,18 @@ func (s *Service) UpdateInvoice(ctx context.Context, input billing.UpdateInvoice
 				return billing.Invoice{}, fmt.Errorf("editing invoice: %w", err)
 			}
 
+			collectionAt := invoice.CollectionAt
+			if ok := UpdateInvoiceCollectionAt(&invoice, billingProfile.Profile.WorkflowConfig.Collection); ok {
+				s.logger.DebugContext(ctx, "collection time updated for invoice",
+					"invoiceID", invoice.ID,
+					"collectionAt", map[string]interface{}{
+						"from":               lo.FromPtr(collectionAt),
+						"to":                 lo.FromPtr(invoice.CollectionAt),
+						"collectionInterval": billingProfile.Profile.WorkflowConfig.Collection.Interval.String(),
+					},
+				)
+			}
+
 			if err := invoice.Validate(); err != nil {
 				return billing.Invoice{}, billing.ValidationError{
 					Err: err,

openmeter/billing/service/invoiceline.go

@@ -45,7 +45,14 @@ func (s *Service) CreatePendingInvoiceLines(ctx context.Context, input billing.C
 
 	return transaction.Run(ctx, s.adapter, func(ctx context.Context) ([]*billing.Line, error) {
 		out := make([]*billing.Line, 0, len(input.Lines))
-		newInvoiceIDs := []string{}
+
+		type UpsertedInvoiceWithProfile struct {
+			InvoiceID       billing.InvoiceID
+			CustomerProfile *billing.ProfileWithCustomerDetails
+			IsInvoiceNew    bool
+		}
+
+		upsertedInvoices := make(map[billing.InvoiceID]UpsertedInvoiceWithProfile)
 
 		for customerID, lineByCustomer := range createByCustomerID {
 			if err := s.validateCustomerForUpdate(ctx, customerentity.CustomerID{
@@ -81,8 +88,25 @@ func (s *Service) CreatePendingInvoiceLines(ctx context.Context, input billing.C
 							return nil, fmt.Errorf("upserting line[%d]: %w", i, err)
 						}
 
-						if updateResult.IsInvoiceNew {
-							newInvoiceIDs = append(newInvoiceIDs, updateResult.Invoice.ID)
+						if updateResult.Invoice == nil {
+							return nil, fmt.Errorf("invoice not found for line[%d]", i)
+						}
+
+						invoiceID := billing.InvoiceID{
+							Namespace: input.Namespace,
+							ID:        updateResult.Invoice.ID,
+						}
+
+						// NOTE: invoices collected once at first encounter in order to make sure that the information
+						// (captured by IsInvoiceNew attribute) about newly created gathering invoices is kept.
+						// Subsequent updates would shadow this parameter which would prevent the system to send system events
+						// about newly created invoices.
+						if _, ok := upsertedInvoices[invoiceID]; !ok {
+							upsertedInvoices[invoiceID] = UpsertedInvoiceWithProfile{
+								InvoiceID:       invoiceID,
+								CustomerProfile: customerProfile,
+								IsInvoiceNew:    updateResult.IsInvoiceNew,
+							}
 						}
 
 						lineService, err := s.lineService.FromEntity(&updateResult.Line)
@@ -120,20 +144,37 @@ func (s *Service) CreatePendingInvoiceLines(ctx context.Context, input billing.C
 			out = append(out, createdLines...)
 		}
 
-		for _, invoiceID := range newInvoiceIDs {
+		for _, upsertedInvoice := range upsertedInvoices {
 			invoice, err := s.GetInvoiceByID(ctx, billing.GetInvoiceByIdInput{
-				Invoice: billing.InvoiceID{
-					Namespace: input.Namespace,
-					ID:        invoiceID,
-				},
-				Expand: billing.InvoiceExpandAll,
+				Invoice: upsertedInvoice.InvoiceID,
+				Expand:  billing.InvoiceExpandAll,
 			})
 			if err != nil {
-				return nil, fmt.Errorf("fetching invoice[%s]: %w", invoiceID, err)
+				return nil, fmt.Errorf("fetching invoice[%s]: %w", upsertedInvoice.InvoiceID.ID, err)
+			}
+
+			// Update invoice if collectionAt field has changed
+			collectionConfig := upsertedInvoice.CustomerProfile.Profile.WorkflowConfig.Collection
+			collectionAt := invoice.CollectionAt
+			if ok := UpdateInvoiceCollectionAt(&invoice, collectionConfig); ok {
+				s.logger.DebugContext(ctx, "collection time updated for invoice",
+					"invoiceID", invoice.ID,
+					"collectionAt", map[string]interface{}{
+						"from":               lo.FromPtr(collectionAt),
+						"to":                 lo.FromPtr(invoice.CollectionAt),
+						"collectionInterval": collectionConfig.Interval.String(),
+					},
+				)
+				if _, err = s.adapter.UpdateInvoice(ctx, invoice); err != nil {
+					return nil, fmt.Errorf("failed to update invoice[%s]: %w", upsertedInvoice.InvoiceID.ID, err)
+				}
 			}
 
-			if err := s.publisher.Publish(ctx, billing.NewInvoiceCreatedEvent(invoice)); err != nil {
-				return nil, fmt.Errorf("publishing invoice[%s] created event: %w", invoiceID, err)
+			// Publish system event for newly created invoices
+			if upsertedInvoice.IsInvoiceNew {
+				if err := s.publisher.Publish(ctx, billing.NewInvoiceCreatedEvent(invoice)); err != nil {
+					return nil, fmt.Errorf("publishing invoice[%s] created event: %w", upsertedInvoice.InvoiceID.ID, err)
+				}
 			}
 		}