fix: transaction handling (#3369)

Author: chrisgacsal

openmeter/app/stripe/adapter/customer.go

@@ -112,7 +112,7 @@ func (a *adapter) UpsertStripeCustomerData(ctx context.Context, input appstripee
 	// Start transaction
 	_, err = entutils.TransactingRepo(ctx, a, func(ctx context.Context, repo *adapter) (any, error) {
 		// Make sure the customer has an app relationship
-		err := a.appService.EnsureCustomer(ctx, app.EnsureCustomerInput{
+		err := repo.appService.EnsureCustomer(ctx, app.EnsureCustomerInput{
 			AppID:      input.AppID,
 			CustomerID: input.CustomerID,
 		})
@@ -207,7 +207,7 @@ func (a *adapter) DeleteStripeCustomerData(ctx context.Context, input appstripee
 		}
 
 		// Delete app customer relationship
-		err = a.appService.DeleteCustomer(ctx, app.DeleteCustomerInput{
+		err = repo.appService.DeleteCustomer(ctx, app.DeleteCustomerInput{
 			AppID:      input.AppID,
 			CustomerID: input.CustomerID,
 		})

openmeter/billing/adapter/customeroverride.go

@@ -113,7 +113,7 @@ func (a *adapter) UpdateCustomerOverride(ctx context.Context, input billing.Upda
 
 func (a *adapter) GetCustomerOverride(ctx context.Context, input billing.GetCustomerOverrideAdapterInput) (*billing.CustomerOverride, error) {
 	return entutils.TransactingRepo(ctx, a, func(ctx context.Context, tx *adapter) (*billing.CustomerOverride, error) {
-		query := a.db.BillingCustomerOverride.Query().
+		query := tx.db.BillingCustomerOverride.Query().
 			Where(billingcustomeroverride.Namespace(input.Customer.Namespace)).
 			Where(billingcustomeroverride.CustomerID(input.Customer.ID)).
 			WithBillingProfile(func(bpq *db.BillingProfileQuery) {
@@ -135,7 +135,7 @@ func (a *adapter) GetCustomerOverride(ctx context.Context, input billing.GetCust
 
 		if dbCustomerOverride.BillingProfileID == nil {
 			// Let's fetch the default billing profile
-			dbDefaultProfile, err := a.db.BillingProfile.Query().
+			dbDefaultProfile, err := tx.db.BillingProfile.Query().
 				Where(billingprofile.Namespace(input.Customer.Namespace)).
 				Where(billingprofile.Default(true)).
 				Where(billingprofile.DeletedAtIsNil()).
@@ -158,7 +158,7 @@ func (a *adapter) ListCustomerOverrides(ctx context.Context, input billing.ListC
 	// Warning: We need to use the customer db parts as for the UI (and for a good API) we need to
 	// be able to filter based on customer fields too.
 	return entutils.TransactingRepo(ctx, a, func(ctx context.Context, tx *adapter) (billing.ListCustomerOverridesAdapterResult, error) {
-		query := a.db.Customer.Query().
+		query := tx.db.Customer.Query().
 			Where(dbcustomer.NamespaceEQ(input.Namespace)).
 			Where(dbcustomer.DeletedAtIsNil())
 

openmeter/meter/adapter/manage.go

@@ -124,7 +124,7 @@ func (a *Adapter) HasActiveFeatureForMeter(ctx context.Context, namespace, key s
 			ctx,
 			a,
 			func(ctx context.Context, repo *Adapter) (bool, error) {
-				exists, err := a.db.Feature.Query().
+				exists, err := repo.db.Feature.Query().
 					Where(featuredb.Namespace(namespace)).
 					Where(featuredb.MeterSlug(key)).
 					Where(featuredb.Or(featuredb.ArchivedAtIsNil(), featuredb.ArchivedAtGT(clock.Now()))).
@@ -166,7 +166,7 @@ func (a *Adapter) ListFeaturesForMeter(ctx context.Context, namespace, key strin
 			ctx,
 			a,
 			func(ctx context.Context, repo *Adapter) ([]feature.Feature, error) {
-				featureRows, err := a.db.Feature.Query().
+				featureRows, err := repo.db.Feature.Query().
 					Where(featuredb.Namespace(namespace)).
 					Where(featuredb.MeterSlug(key)).
 					Where(featuredb.And(