feat: allow triggering subscription sync via events (#3695)

Author: turip

openmeter/billing/worker/subscription/sync.go

@@ -121,6 +121,19 @@ func (h *Handler) invoicePendingLines(ctx context.Context, customer customer.Cus
 	})
 }
 
+func (h *Handler) HandleSubscriptionSyncEvent(ctx context.Context, event *subscription.SubscriptionSyncEvent) error {
+	if event == nil {
+		return nil
+	}
+
+	subsView, err := h.subscriptionService.GetView(ctx, event.Subscription.NamespacedID)
+	if err != nil {
+		return fmt.Errorf("getting subscription view: %w", err)
+	}
+
+	return h.SyncronizeSubscriptionAndInvoiceCustomer(ctx, subsView, time.Now())
+}
+
 func (h *Handler) SyncronizeSubscriptionAndInvoiceCustomer(ctx context.Context, subs subscription.SubscriptionView, asOf time.Time) error {
 	span := tracex.StartWithNoValue(ctx, h.tracer, "billing.worker.subscription.sync.SynchronizeSubscriptionAndInvoiceCustomer", trace.WithAttributes(
 		attribute.String("subscription_id", subs.Subscription.ID),

openmeter/billing/worker/worker.go

@@ -147,6 +147,9 @@ func (w *Worker) eventHandler(opts WorkerOptions) (*grouphandler.NoPublishingHan
 		grouphandler.NewGroupEventHandler(func(ctx context.Context, event *subscription.UpdatedEvent) error {
 			return w.subscriptionSyncHandler.SyncronizeSubscriptionAndInvoiceCustomer(ctx, event.UpdatedView, time.Now())
 		}),
+		grouphandler.NewGroupEventHandler(func(ctx context.Context, event *subscription.SubscriptionSyncEvent) error {
+			return w.subscriptionSyncHandler.HandleSubscriptionSyncEvent(ctx, event)
+		}),
 		grouphandler.NewGroupEventHandler(func(ctx context.Context, event *billing.AdvanceInvoiceEvent) error {
 			return w.asyncAdvanceHandler.Handle(ctx, event)
 		}),

openmeter/subscription/events.go

@@ -2,6 +2,7 @@ package subscription
 
 import (
 	"context"
+	"errors"
 
 	"github.com/openmeterio/openmeter/openmeter/event/metadata"
 	"github.com/openmeterio/openmeter/openmeter/session"
@@ -194,3 +195,48 @@ func (s UpdatedEvent) EventMetadata() metadata.EventMetadata {
 func (s UpdatedEvent) Validate() error {
 	return s.UpdatedView.Validate(true)
 }
+
+type SubscriptionSyncEvent struct {
+	Subscription Subscription `json:"subscription"`
+}
+
+func NewSubscriptionSyncEvent(ctx context.Context, sub Subscription) SubscriptionSyncEvent {
+	return SubscriptionSyncEvent{
+		Subscription: sub,
+	}
+}
+
+var (
+	_ marshaler.Event = SubscriptionSyncEvent{}
+
+	subscriptionSyncEventName = metadata.GetEventName(metadata.EventType{
+		Subsystem: EventSubsystem,
+		Name:      "subscription.sync",
+		Version:   "v1",
+	})
+)
+
+func (s SubscriptionSyncEvent) EventName() string {
+	return subscriptionSyncEventName
+}
+
+func (s SubscriptionSyncEvent) Validate() error {
+	var errs []error
+
+	if err := s.Subscription.NamespacedID.Validate(); err != nil {
+		errs = append(errs, err)
+	}
+
+	if s.Subscription.CustomerId == "" {
+		errs = append(errs, errors.New("customer id is required"))
+	}
+
+	return errors.Join(errs...)
+}
+
+func (s SubscriptionSyncEvent) EventMetadata() metadata.EventMetadata {
+	return metadata.EventMetadata{
+		Source:  metadata.ComposeResourcePath(s.Subscription.Namespace, metadata.EntitySubscription, s.Subscription.ID),
+		Subject: metadata.ComposeResourcePath(s.Subscription.Namespace, metadata.EntityCustomer, s.Subscription.CustomerId),
+	}
+}