File size: 3,150 Bytes
1c4c66b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | 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)
})
}
|