billing/worker/subscriptionsync/adapter/syncstate.go

package adapter

import (
	"context"

	"github.com/samber/lo"

	"github.com/openmeterio/openmeter/openmeter/billing/worker/subscriptionsync"
	entdb "github.com/openmeterio/openmeter/openmeter/ent/db"
	"github.com/openmeterio/openmeter/openmeter/ent/db/predicate"
	"github.com/openmeterio/openmeter/openmeter/ent/db/subscriptionbillingsyncstate"
	"github.com/openmeterio/openmeter/pkg/framework/entutils"
	"github.com/openmeterio/openmeter/pkg/models"
)

var _ subscriptionsync.SyncStateAdapter = (*adapter)(nil)

func (a *adapter) InvalidateSyncState(ctx context.Context, input subscriptionsync.InvalidateSyncStateInput) error {
	return entutils.TransactingRepoWithNoValue(ctx, a, func(ctx context.Context, tx *adapter) error {
		_, err := tx.db.SubscriptionBillingSyncState.Delete().
			Where(subscriptionbillingsyncstate.SubscriptionID(input.ID)).
			Where(subscriptionbillingsyncstate.Namespace(input.Namespace)).
			Exec(ctx)

		return err
	})
}

func (a *adapter) GetSyncStates(ctx context.Context, input subscriptionsync.GetSyncStatesInput) ([]subscriptionsync.SyncState, error) {
	return entutils.TransactingRepo(ctx, a, func(ctx context.Context, tx *adapter) ([]subscriptionsync.SyncState, error) {
		res, err := tx.db.SubscriptionBillingSyncState.Query().
			Where(
				subscriptionbillingsyncstate.Or(
					lo.Map(input, func(id models.NamespacedID, _ int) predicate.SubscriptionBillingSyncState {
						return subscriptionbillingsyncstate.And(
							subscriptionbillingsyncstate.SubscriptionID(id.ID),
							subscriptionbillingsyncstate.Namespace(id.Namespace),
						)
					})...),
			).All(ctx)
		if err != nil {
			return nil, err
		}

		return lo.Map(res, func(state *entdb.SubscriptionBillingSyncState, _ int) subscriptionsync.SyncState {
			return mapSyncStateFromDB(state)
		}), nil
	})
}

func mapSyncStateFromDB(state *entdb.SubscriptionBillingSyncState) subscriptionsync.SyncState {
	nextSyncAfter := state.NextSyncAfter
	if nextSyncAfter != nil {
		nextSyncAfter = lo.ToPtr(nextSyncAfter.UTC())
	}

	return subscriptionsync.SyncState{
		SubscriptionID: models.NamespacedID{ID: state.SubscriptionID, Namespace: state.Namespace},
		HasBillables:   state.HasBillables,
		SyncedAt:       state.SyncedAt.UTC(),
		NextSyncAfter:  nextSyncAfter,
	}
}

func (a *adapter) UpsertSyncState(ctx context.Context, input subscriptionsync.UpsertSyncStateInput) error {
	if err := input.Validate(); err != nil {
		return err
	}

	return entutils.TransactingRepoWithNoValue(ctx, a, func(ctx context.Context, tx *adapter) error {
		nextSyncAfter := input.NextSyncAfter
		if nextSyncAfter != nil {
			nextSyncAfter = lo.ToPtr(nextSyncAfter.UTC())
		}

		return tx.db.SubscriptionBillingSyncState.Create().
			SetHasBillables(input.HasBillables).
			SetSyncedAt(input.SyncedAt.UTC()).
			SetNillableNextSyncAfter(nextSyncAfter).
			SetSubscriptionID(input.SubscriptionID.ID).
			SetNamespace(input.SubscriptionID.Namespace).
			OnConflictColumns(
				subscriptionbillingsyncstate.FieldSubscriptionID,
				subscriptionbillingsyncstate.FieldNamespace,
			).
			UpdateHasBillables().
			UpdateSyncedAt().
			UpdateNextSyncAfter().
			Exec(ctx)
	})
}