feat: add async invoice collection event (#4625)

Author: turip

app/common/openmeter_billingworker.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/openmeterio/openmeter/app/config"
 	billingworker "github.com/openmeterio/openmeter/openmeter/billing/worker"
+	billingworkercollect "github.com/openmeterio/openmeter/openmeter/billing/worker/collect"
 	"github.com/openmeterio/openmeter/openmeter/billing/worker/subscriptionsync"
 	watermillkafka "github.com/openmeterio/openmeter/openmeter/watermill/driver/kafka"
 	"github.com/openmeterio/openmeter/openmeter/watermill/eventbus"
@@ -38,6 +39,7 @@ var BillingWorker = wire.NewSet(
 
 	NewBillingWorkerOptions,
 	NewBillingWorker,
+	NewBillingCollector,
 	NewBillingSubscriptionSyncAdapter,
 	NewBillingSubscriptionSyncService,
 	BillingWorkerGroup,
@@ -77,6 +79,7 @@ func NewBillingWorkerOptions(
 	billingRegistry BillingRegistry,
 	subscriptionServices SubscriptionServiceWithWorkflow,
 	subscriptionSyncService subscriptionsync.Service,
+	billingCollector *billingworkercollect.InvoiceCollector,
 	billingFsConfig config.BillingFeatureSwitchesConfiguration,
 	logger *slog.Logger,
 ) billingworker.WorkerOptions {
@@ -86,6 +89,7 @@ func NewBillingWorkerOptions(
 		Router:                  routerOptions,
 		EventBus:                eventBus,
 		BillingService:          billingRegistry.Billing,
+		BillingCollector:        billingCollector,
 		ChargesService:          billingRegistry.ChargesServiceOrNil(),
 		SubscriptionService:     subscriptionServices.Service,
 		BillingSubscriptionSync: subscriptionSyncService,

cmd/billing-worker/wire_gen.go

@@ -0,0 +1,49 @@
+package billing
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/openmeterio/openmeter/openmeter/event/metadata"
+	"github.com/openmeterio/openmeter/pkg/models"
+)
+
+type CollectCustomerInvoicesEvent struct {
+	Namespace  string    `json:"namespace"`
+	CustomerID string    `json:"customer_id"`
+	AsOf       time.Time `json:"as_of"`
+}
+
+func (e CollectCustomerInvoicesEvent) EventName() string {
+	return metadata.GetEventName(metadata.EventType{
+		Subsystem: EventSubsystem,
+		Name:      "invoice.collect",
+		Version:   "v1",
+	})
+}
+
+func (e CollectCustomerInvoicesEvent) EventMetadata() metadata.EventMetadata {
+	return metadata.EventMetadata{
+		Source:  metadata.ComposeResourcePath(e.Namespace, metadata.EntityCustomer, e.CustomerID),
+		Subject: metadata.ComposeResourcePath(e.Namespace, metadata.EntityCustomer, e.CustomerID),
+	}
+}
+
+func (e CollectCustomerInvoicesEvent) Validate() error {
+	var errs []error
+
+	if e.Namespace == "" {
+		errs = append(errs, fmt.Errorf("namespace cannot be empty"))
+	}
+
+	if e.CustomerID == "" {
+		errs = append(errs, fmt.Errorf("customer_id cannot be empty"))
+	}
+
+	if e.AsOf.IsZero() {
+		errs = append(errs, fmt.Errorf("as_of cannot be zero"))
+	}
+
+	return models.NewNillableGenericValidationError(errors.Join(errs...))
+}

openmeter/billing/eventscollection.go

@@ -0,0 +1,73 @@
+package billing_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/openmeterio/openmeter/openmeter/billing"
+	"github.com/openmeterio/openmeter/openmeter/event/metadata"
+)
+
+func TestCollectCustomerInvoicesEvent(t *testing.T) {
+	asOf := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC)
+
+	event := billing.CollectCustomerInvoicesEvent{
+		Namespace:  "ns",
+		CustomerID: "customer-id",
+		AsOf:       asOf,
+	}
+
+	require.NoError(t, event.Validate())
+	require.Equal(t, metadata.GetEventName(metadata.EventType{
+		Subsystem: billing.EventSubsystem,
+		Name:      "invoice.collect",
+		Version:   "v1",
+	}), event.EventName())
+
+	eventMetadata := event.EventMetadata()
+	require.Equal(t, metadata.ComposeResourcePath("ns", metadata.EntityCustomer, "customer-id"), eventMetadata.Source)
+	require.Equal(t, metadata.ComposeResourcePath("ns", metadata.EntityCustomer, "customer-id"), eventMetadata.Subject)
+}
+
+func TestCollectCustomerInvoicesEventValidate(t *testing.T) {
+	asOf := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC)
+
+	tests := []struct {
+		name    string
+		event   billing.CollectCustomerInvoicesEvent
+		wantErr string
+	}{
+		{
+			name: "namespace is required",
+			event: billing.CollectCustomerInvoicesEvent{
+				CustomerID: "customer-id",
+				AsOf:       asOf,
+			},
+			wantErr: "namespace cannot be empty",
+		},
+		{
+			name: "customer id is required",
+			event: billing.CollectCustomerInvoicesEvent{
+				Namespace: "ns",
+				AsOf:      asOf,
+			},
+			wantErr: "customer_id cannot be empty",
+		},
+		{
+			name: "as of is required",
+			event: billing.CollectCustomerInvoicesEvent{
+				Namespace:  "ns",
+				CustomerID: "customer-id",
+			},
+			wantErr: "as_of cannot be zero",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			require.ErrorContains(t, tt.event.Validate(), tt.wantErr)
+		})
+	}
+}

openmeter/billing/eventscollection_test.go

@@ -0,0 +1,29 @@
+package billingworkercollect
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/openmeterio/openmeter/openmeter/billing"
+	"github.com/openmeterio/openmeter/openmeter/customer"
+)
+
+func (a *InvoiceCollector) HandleCollectCustomerInvoicesEvent(ctx context.Context, event *billing.CollectCustomerInvoicesEvent) error {
+	if event == nil {
+		return nil
+	}
+
+	if err := event.Validate(); err != nil {
+		return fmt.Errorf("invalid collect customer invoices event: %w", err)
+	}
+
+	_, err := a.CollectCustomerInvoice(ctx, CollectCustomerInvoiceInput{
+		CustomerID: customer.CustomerID{
+			Namespace: event.Namespace,
+			ID:        event.CustomerID,
+		},
+		AsOf: event.AsOf,
+	})
+
+	return err
+}

openmeter/billing/worker/collect/asynccollect.go

@@ -0,0 +1,27 @@
+package billingworkercollect
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/openmeterio/openmeter/openmeter/billing"
+)
+
+func TestInvoiceCollectorHandleCollectCustomerInvoicesEvent(t *testing.T) {
+	collector := &InvoiceCollector{}
+
+	t.Run("nil event", func(t *testing.T) {
+		require.NoError(t, collector.HandleCollectCustomerInvoicesEvent(t.Context(), nil))
+	})
+
+	t.Run("invalid event", func(t *testing.T) {
+		err := collector.HandleCollectCustomerInvoicesEvent(t.Context(), &billing.CollectCustomerInvoicesEvent{
+			Namespace:  "ns",
+			CustomerID: "customer-id",
+		})
+
+		require.ErrorContains(t, err, "invalid collect customer invoices event")
+		require.ErrorContains(t, err, "as_of cannot be zero")
+	})
+}

openmeter/billing/worker/collect/asynccollect_test.go

@@ -106,6 +106,7 @@ func (a *InvoiceCollector) CollectCustomerInvoice(ctx context.Context, params Co
 		ctx,
 		billing.InvoicePendingLinesInput{
 			Customer: params.CustomerID,
+			AsOf:     lo.ToPtr(params.AsOf),
 		},
 		// We want to make sure that system collection does not use progressive billing.
 		billing.WithPartialInvoiceLinesDisabled(),

openmeter/billing/worker/collect/collect.go

@@ -13,6 +13,7 @@ import (
 	"github.com/openmeterio/openmeter/openmeter/billing/charges"
 	chargesasyncadvance "github.com/openmeterio/openmeter/openmeter/billing/charges/worker/asyncadvance"
 	"github.com/openmeterio/openmeter/openmeter/billing/worker/asyncadvance"
+	billingworkercollect "github.com/openmeterio/openmeter/openmeter/billing/worker/collect"
 	"github.com/openmeterio/openmeter/openmeter/billing/worker/subscriptionsync"
 	"github.com/openmeterio/openmeter/openmeter/subscription"
 	"github.com/openmeterio/openmeter/openmeter/watermill/eventbus"
@@ -30,6 +31,7 @@ type WorkerOptions struct {
 
 	BillingService          billing.Service
 	BillingSubscriptionSync subscriptionsync.Service
+	BillingCollector        *billingworkercollect.InvoiceCollector
 	// ChargesService is optional; when non-nil the worker handles AdvanceChargesEvent.
 	ChargesService charges.ChargeService
 	// External connectors
@@ -67,6 +69,10 @@ func (w WorkerOptions) Validate() error {
 		return fmt.Errorf("billing subscription sync handler is required")
 	}
 
+	if w.BillingCollector == nil {
+		return fmt.Errorf("billing collector is required")
+	}
+
 	return nil
 }
 
@@ -75,6 +81,7 @@ type Worker struct {
 
 	billingService             billing.Service
 	subscriptionSync           subscriptionsync.Service
+	invoiceCollector           *billingworkercollect.InvoiceCollector
 	asyncAdvanceHandler        *asyncadvance.Handler
 	asyncAdvanceChargesHandler *chargesasyncadvance.Handler
 	nonPublishingHandler       *grouphandler.NoPublishingHandler
@@ -109,6 +116,7 @@ func New(opts WorkerOptions) (*Worker, error) {
 	worker := &Worker{
 		billingService:             opts.BillingService,
 		subscriptionSync:           opts.BillingSubscriptionSync,
+		invoiceCollector:           opts.BillingCollector,
 		asyncAdvanceHandler:        asyncAdvancer,
 		asyncAdvanceChargesHandler: asyncAdvanceChargesHandler,
 		lockdownNamespaces:         opts.LockdownNamespaces,
@@ -199,6 +207,13 @@ func (w *Worker) eventHandler(opts WorkerOptions) (*grouphandler.NoPublishingHan
 
 			return w.asyncAdvanceHandler.Handle(ctx, event)
 		}),
+		grouphandler.NewGroupEventHandler(func(ctx context.Context, event *billing.CollectCustomerInvoicesEvent) error {
+			if event != nil && slices.Contains(w.lockdownNamespaces, event.Namespace) {
+				return nil
+			}
+
+			return w.invoiceCollector.HandleCollectCustomerInvoicesEvent(ctx, event)
+		}),
 		grouphandler.NewGroupEventHandler(func(ctx context.Context, event *charges.AdvanceChargesEvent) error {
 			if w.asyncAdvanceChargesHandler == nil {
 				return nil