billing/worker/subscriptionsync/service/service.go

package service

import (
	"context"
	"fmt"
	"log/slog"
	"time"

	"go.opentelemetry.io/otel/trace"

	"github.com/openmeterio/openmeter/openmeter/billing"
	"github.com/openmeterio/openmeter/openmeter/billing/charges"
	"github.com/openmeterio/openmeter/openmeter/billing/worker/subscriptionsync"
	"github.com/openmeterio/openmeter/openmeter/billing/worker/subscriptionsync/service/reconciler"
	"github.com/openmeterio/openmeter/openmeter/subscription"
	"github.com/openmeterio/openmeter/pkg/featuregate"
	"github.com/openmeterio/openmeter/pkg/framework/transaction"
	"github.com/openmeterio/openmeter/pkg/models"
)

type FeatureFlags struct {
	EnableFlatFeeInAdvanceProrating bool
	EnableFlatFeeInArrearsProrating bool
	EnableCreditThenInvoice         bool
	MaxLinesPerCollectedInvoice     int
}

type Config struct {
	BillingService billing.Service
	// ChargesService is required for credit-only sync and charge-based provisioning.
	ChargesService                charges.Service
	SubscriptionService           subscription.Service
	SubscriptionSyncAdapter       subscriptionsync.Adapter
	FeatureFlags                  FeatureFlags
	ForceAsyncInvoicePendingLines bool
	Logger                        *slog.Logger
	Tracer                        trace.Tracer
	FeatureGate                   *featuregate.FeatureGateChecker
}

func (c Config) Validate() error {
	if c.BillingService == nil {
		return fmt.Errorf("billing service is required")
	}

	if c.SubscriptionService == nil {
		return fmt.Errorf("subscription service is required")
	}

	if c.SubscriptionSyncAdapter == nil {
		return fmt.Errorf("subscription sync adapter is required")
	}

	if c.Logger == nil {
		return fmt.Errorf("logger is required")
	}

	if c.Tracer == nil {
		return fmt.Errorf("tracer is required")
	}

	if err := c.FeatureGate.Validate(); err != nil {
		return err
	}

	return nil
}

var _ subscriptionsync.Service = (*Service)(nil)

type Service struct {
	billingService                billing.Service
	chargesService                charges.Service
	reconciler                    reconciler.Reconciler
	subscriptionService           subscription.Service
	subscriptionSyncAdapter       subscriptionsync.Adapter
	featureFlags                  FeatureFlags
	forceAsyncInvoicePendingLines bool
	logger                        *slog.Logger
	tracer                        trace.Tracer
}

func New(config Config) (*Service, error) {
	if err := config.Validate(); err != nil {
		return nil, err
	}
	reconcilerSvc, err := reconciler.New(reconciler.Config{
		BillingService:          config.BillingService,
		ChargesService:          config.ChargesService,
		EnableCreditThenInvoice: config.FeatureFlags.EnableCreditThenInvoice,
		Logger:                  config.Logger,
		FeatureGate:             config.FeatureGate,
	})
	if err != nil {
		return nil, err
	}
	return &Service{
		billingService:                config.BillingService,
		chargesService:                config.ChargesService,
		reconciler:                    reconcilerSvc,
		subscriptionSyncAdapter:       config.SubscriptionSyncAdapter,
		featureFlags:                  config.FeatureFlags,
		forceAsyncInvoicePendingLines: config.ForceAsyncInvoicePendingLines,
		subscriptionService:           config.SubscriptionService,
		logger:                        config.Logger,
		tracer:                        config.Tracer,
	}, nil
}

func (s *Service) GetSyncStates(ctx context.Context, input subscriptionsync.GetSyncStatesInput) ([]subscriptionsync.SyncState, error) {
	return transaction.Run(ctx, s.subscriptionSyncAdapter, func(ctx context.Context) ([]subscriptionsync.SyncState, error) {
		return s.subscriptionSyncAdapter.GetSyncStates(ctx, input)
	})
}

func (s *Service) SyncByViewAndInvoiceCustomer(ctx context.Context, view subscription.SubscriptionView, asOf time.Time) error {
	return s.synchronizeSubscriptionAndInvoiceCustomer(ctx, newSubscriptionReferenceOrView(view), asOf)
}

func (s *Service) SyncByIDAndInvoiceCustomer(ctx context.Context, subscriptionID models.NamespacedID, asOf time.Time) error {
	return s.synchronizeSubscriptionAndInvoiceCustomer(ctx, newSubscriptionReferenceOrView(subscriptionID), asOf)
}

func (s *Service) SyncByView(ctx context.Context, view subscription.SubscriptionView, asOf time.Time, opts ...subscriptionsync.SynchronizeSubscriptionOption) error {
	_, err := s.synchronizeSubscription(ctx, newSubscriptionReferenceOrView(view), asOf, opts...)
	return err
}

func (s *Service) SyncByID(ctx context.Context, subscriptionID models.NamespacedID, asOf time.Time, opts ...subscriptionsync.SynchronizeSubscriptionOption) error {
	_, err := s.synchronizeSubscription(ctx, newSubscriptionReferenceOrView(subscriptionID), asOf, opts...)
	return err
}