subscription/service/sync.go

package service

import (
	"context"
	"fmt"

	"github.com/samber/lo"

	"github.com/openmeterio/openmeter/openmeter/entitlement"
	"github.com/openmeterio/openmeter/openmeter/subscription"
	"github.com/openmeterio/openmeter/pkg/convert"
	"github.com/openmeterio/openmeter/pkg/framework/transaction"
	"github.com/openmeterio/openmeter/pkg/models"
)

// Sync manages the synchronization of a subscription with a new spec.
// It consists of 3 steps:
// 1. Remove anything that's changed or got removed
// 2. Create anything that's been changed
// 3. Create anything that's new
//
// Some remarks:
// 1. Change comparison is done on the relevant create inputs.
// 2. Things being deleted are marked via a `touched` map.
//
// TODO: localize error so phase and item keys are always included (alongside subscription reference)
// TODO (OM-1074): clean up this control flow
func (s *service) sync(ctx context.Context, view subscription.SubscriptionView, newSpec subscription.SubscriptionSpec) (subscription.Subscription, error) {
	return transaction.Run(ctx, s.TransactionManager, func(ctx context.Context) (subscription.Subscription, error) {
		var def subscription.Subscription

		// Some sanity checks for good measure
		if view.Subscription.CustomerId != newSpec.CustomerId {
			return def, fmt.Errorf("cannot change customer id")
		}
		if !view.Subscription.PlanRef.NilEqual(newSpec.Plan) {
			return def, fmt.Errorf("cannot change plan")
		}
		if !view.Subscription.ActiveFrom.Equal(newSpec.ActiveFrom) {
			return def, fmt.Errorf("cannot change subscription start")
		}
		if view.Subscription.SettlementMode != newSpec.SettlementMode {
			return def, fmt.Errorf("cannot change settlement mode")
		}

		dirty := make(touched)

		// Let's make sure the Subscription Cadence is up to date
		if !view.Subscription.CadencedModel.Equal(models.CadencedModel{ActiveFrom: newSpec.ActiveFrom, ActiveTo: newSpec.ActiveTo}) {
			_, err := s.SubscriptionRepo.SetEndOfCadence(ctx, view.Subscription.NamespacedID, newSpec.ActiveTo)
			if err != nil {
				return def, fmt.Errorf("failed to set end of cadence: %w", err)
			}
		}

		// 1. Let's remove anything that's changed or got removed
		newSortedPhaseSpecs := newSpec.GetSortedPhases()
		for _, currentPhaseView := range view.Phases {
			// Let's try find a matching phase in the new spec
			matchingPhaseFromNewSpec, found := lo.Find(newSortedPhaseSpecs, func(s *subscription.SubscriptionPhaseSpec) bool {
				return s.PhaseKey == currentPhaseView.SubscriptionPhase.Key
			})

			// If there's no equivalent found in the current spec, we need to delete the phase
			if !found {
				if err := s.deletePhase(ctx, currentPhaseView); err != nil {
					return def, fmt.Errorf("failed to delete phase: %w", err)
				}

				dirty.mark(subscription.NewPhasePath(currentPhaseView.SubscriptionPhase.Key))

				// There's nothing more to be done for this phase, so lets skip to the next one
				continue
			}

			// sanity check
			if matchingPhaseFromNewSpec == nil {
				return def, fmt.Errorf("failed to find matching phase in new spec but no error was returned")
			}

			// Let's get the cadence of the current phase
			cadenceOfCurrentPhaseBasedOnSpec, err := view.Spec.GetPhaseCadence(currentPhaseView.SubscriptionPhase.Key)
			if err != nil {
				return def, fmt.Errorf("failed to get cadence for current phase %s: %w", currentPhaseView.SubscriptionPhase.Key, err)
			}

			// Let's get the cadence of the new phase
			cadenceOfNewPhaseBasedOnSpec, err := newSpec.GetPhaseCadence(matchingPhaseFromNewSpec.PhaseKey)
			if err != nil {
				return def, fmt.Errorf("failed to get cadence for new phase %s: %w", matchingPhaseFromNewSpec.PhaseKey, err)
			}

			// Lets figure out when the new phase should start
			newPhaseStartTime, _ := matchingPhaseFromNewSpec.StartAfter.AddTo(view.Subscription.ActiveFrom)

			curr := currentPhaseView.Spec.ToCreateSubscriptionPhaseEntityInput(view.Subscription, currentPhaseView.SubscriptionPhase.ActiveFrom)
			new := matchingPhaseFromNewSpec.ToCreateSubscriptionPhaseEntityInput(view.Subscription, newPhaseStartTime)

			// If the phase has any changes, we need to recreate it. That means also all sub-resources of it have to be relinked.
			if !curr.Equal(new) {
				// This means deleting the phase with all its sub-resources
				if err := s.deletePhase(ctx, currentPhaseView); err != nil {
					return def, fmt.Errorf("failed to delete phase: %w", err)
				}

				dirty.mark(subscription.NewPhasePath(currentPhaseView.SubscriptionPhase.Key))

				// The phase is deleted, there's nothing more to be done
				continue
			}

			// Sanity check, the current phase cannot be dirty
			if dirty.isTouched(subscription.NewPhasePath(currentPhaseView.SubscriptionPhase.Key)) {
				return def, fmt.Errorf("current phase is dirty but should not be")
			}

			// Now let's iterate through all items in the phase
			for currentItemViewsKey, currentItemViews := range currentPhaseView.ItemsByKey {
				// Let's try find a matching item in the new spec
				// Here as we do an update, we rely on the previously verified integrity of both view and spec
				// Due to this, we use a simple matching based on the index of the item under the given key
				matchingItemsByKeyFromNewSpec, found := matchingPhaseFromNewSpec.ItemsByKey[currentItemViewsKey]

				// If no matching key is found in the new spec lets delete everything
				if !found || len(matchingItemsByKeyFromNewSpec) == 0 {
					for _, currentItemView := range currentItemViews {
						if err := s.deleteItem(ctx, currentItemView); err != nil {
							return def, fmt.Errorf("failed to delete item: %w", err)
						}

						dirty.mark(subscription.NewItemPath(currentItemView.Spec.PhaseKey, currentItemView.Spec.ItemKey))
					}

					// There's nothing more to be done for this item(key), so lets skip to the next one
					continue
				}

				for currentItemIdx, currentItemView := range currentItemViews {
					// Let's get the item with the same index from the new spec
					if currentItemIdx >= len(matchingItemsByKeyFromNewSpec) {
						// Let's delete the item as it's not present in the new spec

						if err := s.deleteItem(ctx, currentItemView); err != nil {
							return def, fmt.Errorf("failed to delete item: %w", err)
						}

						dirty.mark(subscription.NewItemVersionPath(currentItemView.Spec.PhaseKey, currentItemView.Spec.ItemKey, currentItemIdx))

						// There's nothing more to be done for this item, so lets skip to the next one
						continue
					}

					matchingItemFromNewSpec := matchingItemsByKeyFromNewSpec[currentItemIdx]

					// First, let's check if the item itself needs to be changed
					curr, err := currentItemView.Spec.ToCreateSubscriptionItemEntityInput(
						currentPhaseView.SubscriptionPhase.NamespacedID,
						cadenceOfCurrentPhaseBasedOnSpec,
						convert.SafeDeRef(currentItemView.Entitlement, func(s subscription.SubscriptionEntitlement) *entitlement.Entitlement {
							return &s.Entitlement.Entitlement
						}),
					)
					if err != nil {
						return def, fmt.Errorf("failed to convert item to entity input: %w", err)
					}

					// Here we don't preamptively know all the properties but fortunately all we need to know is whether they'd change or not
					// We're prepopulating changing fields with invalid values, which is a lie and a bad method, but it's necessary for now due to the hard linking

					newPhaseID := currentPhaseView.SubscriptionPhase.NamespacedID
					if dirty.isTouched(subscription.NewPhasePath(currentPhaseView.SubscriptionPhase.Key)) {
						newPhaseID = impossibleNamespacedId
					}

					newOnlyForComparisonWithInvalidProperties, err := matchingItemFromNewSpec.ToCreateSubscriptionItemEntityInput(
						newPhaseID,
						cadenceOfNewPhaseBasedOnSpec,
						// Without the "new" entitlement already present we cannot properly compare the two create inputs.
						// To work around this, we'll reuse the current entitlement for comparison.
						// This won't cause an issue as all relevant properties of the entitlement are specced on the Item (as part of RateCard)
						// FIXME: This is a lie
						convert.SafeDeRef(currentItemView.Entitlement, func(s subscription.SubscriptionEntitlement) *entitlement.Entitlement {
							return &s.Entitlement.Entitlement
						}),
					)
					if err != nil {
						return def, fmt.Errorf("failed to convert item to entity input: %w", err)
					}

					doesItemNeedToBeChanged := !curr.Equal(newOnlyForComparisonWithInvalidProperties)

					if doesItemNeedToBeChanged {
						// This means deleting the item with all its sub-resources
						if err := s.deleteItem(ctx, currentItemView); err != nil {
							return def, fmt.Errorf("failed to delete item: %w", err)
						}

						dirty.mark(subscription.NewItemVersionPath(currentItemView.Spec.PhaseKey, currentItemView.Spec.ItemKey, currentItemIdx))

						// There's nothing more to be done here, so lets skip to the next one
						continue
					}
				}
			}
		}

		// 2. Let's create anything that's been changed
		for _, currentPhaseView := range view.Phases {
			// Let's try find a matching phase in the new spec
			matchingPhaseFromNewSpec, found := lo.Find(newSortedPhaseSpecs, func(s *subscription.SubscriptionPhaseSpec) bool {
				return s.PhaseKey == currentPhaseView.SubscriptionPhase.Key
			})

			if !found {
				// If the phase wasn't found there's nothing to create
				continue
			}

			// Sanity check
			if matchingPhaseFromNewSpec == nil {
				return def, fmt.Errorf("failed to find matching phase in new spec but no error was returned")
			}

			newPhaseCadence, err := newSpec.GetPhaseCadence(matchingPhaseFromNewSpec.PhaseKey)
			if err != nil {
				return def, fmt.Errorf("failed to get cadence for phase %s: %w", matchingPhaseFromNewSpec.PhaseKey, err)
			}

			// If the phase got deleted, we can create it as a whole
			if dirty.isTouched(subscription.NewPhasePath(currentPhaseView.SubscriptionPhase.Key)) {
				if _, err := s.createPhase(ctx, view.Customer, *matchingPhaseFromNewSpec, view.Subscription, newPhaseCadence); err != nil {
					return def, fmt.Errorf("failed to create phase: %w", err)
				}

				// There's nothing more to be done for this phase, so lets skip to the next one
				continue
			}

			// Now let's check each of the items in the phase
			for currentItemViewsKey, currentItemViews := range currentPhaseView.ItemsByKey {
				// Let's try find a matching item in the new spec
				// Here as we do an update, we rely on the previously verified integrity of both view and spec
				// Due to this, we use a simple matching based on the index of the item under the given key
				matchingItemsByKeyFromNewSpec, found := matchingPhaseFromNewSpec.ItemsByKey[currentItemViewsKey]
				if !found {
					// If the item wasn't found there's nothing to create
					continue
				}

				for currentItemIdx, currentItemView := range currentItemViews {
					// Let's get the item with the same index from the new spec
					if currentItemIdx >= len(matchingItemsByKeyFromNewSpec) {
						// We went out of bounds, these items are not present in the new spec so we can just break

						break
					}

					matchingItemFromNewSpec := matchingItemsByKeyFromNewSpec[currentItemIdx]

					// If the item got deleted, we can create it as a whole
					if dirty.isTouched(subscription.NewItemVersionPath(currentItemView.Spec.PhaseKey, currentItemView.Spec.ItemKey, currentItemIdx)) {
						if _, err := s.createItem(ctx, createItemOptions{
							cust:         view.Customer,
							sub:          view.Subscription,
							phase:        currentPhaseView.SubscriptionPhase,
							phaseCadence: newPhaseCadence,
							itemSpec:     *matchingItemFromNewSpec,
						}); err != nil {
							return def, fmt.Errorf("failed to create item: %w", err)
						}

						// There's nothing more to be done for this item, so lets skip to the next one
						continue
					}
				}
			}
		}

		// 3. Finally, let's create anything that's new
		for _, phase := range newSpec.GetSortedPhases() {
			// Sanity check
			if phase == nil {
				return def, fmt.Errorf("phase is nil")
			}

			// Let's see if the phase was present in the current view
			matchingPhaseInCurrentView, foundMatchingPhaseInCurrentView := lo.Find(view.Phases, func(p subscription.SubscriptionPhaseView) bool {
				return p.SubscriptionPhase.Key == phase.PhaseKey
			})

			if !foundMatchingPhaseInCurrentView {
				phaseCadence, err := newSpec.GetPhaseCadence(phase.PhaseKey)
				if err != nil {
					return def, fmt.Errorf("failed to get cadence for phase %s: %w", phase.PhaseKey, err)
				}

				if _, err := s.createPhase(ctx, view.Customer, *phase, view.Subscription, phaseCadence); err != nil {
					return def, fmt.Errorf("failed to create phase: %w", err)
				}
				continue
			}

			// Now lets check all the items in the phase
			for key, itemsByKey := range phase.ItemsByKey {
				matchingItemsByKeyInCurrentView, foundMatchingItemsByKeyInCurrentView := matchingPhaseInCurrentView.ItemsByKey[key]

				for itemIdx, item := range itemsByKey {
					phaseCadence, err := newSpec.GetPhaseCadence(phase.PhaseKey)
					if err != nil {
						return def, fmt.Errorf("failed to get cadence for phase %s: %w", phase.PhaseKey, err)
					}

					// If we didn't find a matching key in the current view, we need to create the item
					if !foundMatchingItemsByKeyInCurrentView {
						if _, err := s.createItem(ctx, createItemOptions{
							cust:         view.Customer,
							sub:          view.Subscription,
							phase:        matchingPhaseInCurrentView.SubscriptionPhase,
							phaseCadence: phaseCadence,
							itemSpec:     *item,
						}); err != nil {
							return def, fmt.Errorf("failed to create item: %w", err)
						}

						// There's nothing left to do for this item
						continue
					} else if itemIdx >= len(matchingItemsByKeyInCurrentView) {
						// If there's a matching key, then in the previous step we've taken care of all indexes
						// present in the current phase

						// The rest we create
						if _, err := s.createItem(ctx, createItemOptions{
							cust:         view.Customer,
							sub:          view.Subscription,
							phase:        matchingPhaseInCurrentView.SubscriptionPhase,
							phaseCadence: phaseCadence,
							itemSpec:     *item,
						}); err != nil {
							return def, fmt.Errorf("failed to create item: %w", err)
						}
					}
				}
			}
		}

		// 4. Finally we're done with syncing everything, we should just re-fetch the subscription
		return s.Get(ctx, view.Subscription.NamespacedID)
	})
}

// touched is a map of touched paths (honoring sub-resource relationships)
type touched map[subscription.SpecPath]bool

// Mark a given path as touched
func (t touched) mark(key subscription.SpecPath) {
	t[key] = true
}

// Check if a given path has been touched
// If path X has been touched, then all sub-resources of X have been touched
func (t touched) isTouched(key subscription.SpecPath) bool {
	for k := range t {
		// IsParentOf check returns true for identity as well
		if k.IsParentOf(key) {
			return true
		}
	}
	return false
}

// NewItemVersionPath returns an invalid PatchPath thats still usable for IsParentOf checks
// FIXME: this is a hack. For instance, is featureKey were to contain `/` it would completely break (though that exact scenario is otherwise prohibited)

// an ID that can never occur in normal control flow
var impossibleNamespacedId = models.NamespacedID{
	ID:        "impossible",
	Namespace: "impossible",
}