diff --git a/billing/checkout/service.go b/billing/checkout/service.go index ef240b748e..ccda3f493b 100644 --- a/billing/checkout/service.go +++ b/billing/checkout/service.go @@ -788,7 +788,7 @@ func (s *Service) List(ctx context.Context, filter Filter) ([]Checkout, error) { return s.repository.List(ctx, filter) } -// DeleteByCustomer removes all checkout records of a billing account. Checkout +// DeleteByCustomer marks every checkout of a billing account as deleted. Checkout // sessions on the billing provider are not touched as they expire on their own. func (s *Service) DeleteByCustomer(ctx context.Context, customerID string) error { return s.repository.DeleteByCustomerID(ctx, customerID) diff --git a/billing/credit/service.go b/billing/credit/service.go index bd4e4eff38..81396c53d1 100644 --- a/billing/credit/service.go +++ b/billing/credit/service.go @@ -176,7 +176,7 @@ func (s Service) GetByID(ctx context.Context, id string) (Transaction, error) { return s.transactionRepository.GetByID(ctx, id) } -// DeleteByAccountID removes all credit transactions of a billing account. It is +// DeleteByAccountID marks every credit transaction of a billing account as deleted. It is // meant for account teardown. func (s Service) DeleteByAccountID(ctx context.Context, accountID string) error { return s.transactionRepository.DeleteByAccountID(ctx, accountID) diff --git a/billing/customer/errors.go b/billing/customer/errors.go index 60913e50f7..3c417e2b03 100644 --- a/billing/customer/errors.go +++ b/billing/customer/errors.go @@ -13,4 +13,5 @@ var ( ErrInvalidDetail = errors.New("invalid billing customer detail") ErrDisabled = errors.New("billing customer is disabled") ErrExistingAccountWithPendingDues = errors.New("existing account with pending dues found") + ErrHasLiveRecords = errors.New("billing account still has live subscriptions, invoices or checkouts") ) diff --git a/billing/subscription/service.go b/billing/subscription/service.go index 35feee2267..686bb95a60 100644 --- a/billing/subscription/service.go +++ b/billing/subscription/service.go @@ -1102,7 +1102,7 @@ func (s *Service) ensureCreditsForPlan(ctx context.Context, sub Subscription, su // subscriptions are canceled on the billing provider first. A subscription // that is already gone or canceled on the provider counts as canceled, so a // teardown that failed midway can be run again. Offline accounts have nothing -// on the provider; only their local records are removed. +// on the provider; only their local rows are marked deleted. func (s *Service) DeleteByCustomer(ctx context.Context, customr customer.Customer) error { subs, err := s.List(ctx, Filter{ CustomerID: customr.ID, diff --git a/core/deleter/service.go b/core/deleter/service.go index 75ce2a4df6..1e93a73c65 100644 --- a/core/deleter/service.go +++ b/core/deleter/service.go @@ -442,20 +442,10 @@ func (d Service) DeleteCustomers(ctx context.Context, id string) error { // here. func (d Service) deleteCustomers(ctx context.Context, id string, customers []customer.Customer, amounts map[string]accountTokens) error { for _, c := range customers { - // TODO(fix): the subscription and checkout reads here skip rows with - // deleted_at. Once these deletes turn soft, a row left behind with - // deleted_at set is invisible to this loop. A subscription then blocks the - // customer delete on its foreign key, and a checkout is removed without - // its audit record below. Make these deletes soft in the same change. - // cancels active subscriptions on the billing provider and removes local records + // cancels active subscriptions on the billing provider and hides the local rows if err := d.subService.DeleteByCustomer(ctx, c); err != nil { return fmt.Errorf("failed to delete org while deleting a billing account subscriptions[%s]: %w", c.ID, err) } - // TODO(fix): this delete is due to become a soft delete. An invoice will - // only be allowed to carry deleted_at once its customer already does, - // and the customer below is deleted last, so this order gets rejected. - // Delete the customer first, or run the whole loop in one transaction - // with a deferred check. if err := d.invoiceService.DeleteByCustomer(ctx, c); err != nil { return fmt.Errorf("failed to delete org while deleting a billing account invoices[%s]: %w", c.ID, err) } @@ -466,8 +456,8 @@ func (d Service) deleteCustomers(ctx context.Context, id string, customers []cus if err := d.checkoutService.DeleteByCustomer(ctx, c.ID); err != nil { return fmt.Errorf("failed to delete org while deleting a billing account checkouts[%s]: %w", c.ID, err) } - // the checkout rows are gone after this; the audit records keep the - // provider references so the sessions can still be found on the provider. + // the checkout rows are hidden after this; the audit records say which + // sessions the delete covered. // The records are only written when ctx carries the audit service (the // API path seeds it); otherwise audit.NewLogger falls back to a noop auditLogger := audit.NewLogger(ctx, id) @@ -495,7 +485,7 @@ func (d Service) deleteCustomers(ctx context.Context, id string, customers []cus // tokens still on the account are forfeited by this delete, so // record the amount before the transactions are removed. The // purchased share goes on the record too: the transaction rows are - // deleted right after, and support settles a transfer from this + // hidden right after, and support settles a transfer from this // number later account, err := d.resolveAccountTokens(ctx, c.ID, amounts) if err != nil { diff --git a/internal/api/v1beta1connect/billing_errors.go b/internal/api/v1beta1connect/billing_errors.go index 4eccf91e6d..af840537fd 100644 --- a/internal/api/v1beta1connect/billing_errors.go +++ b/internal/api/v1beta1connect/billing_errors.go @@ -69,6 +69,8 @@ func mapBillingErrorCode(err error) *connect.Error { return connect.NewError(connect.CodeFailedPrecondition, subscription.ErrPhaseIsUpdating) case errors.Is(err, customer.ErrExistingAccountWithPendingDues): return connect.NewError(connect.CodeFailedPrecondition, customer.ErrExistingAccountWithPendingDues) + case errors.Is(err, customer.ErrHasLiveRecords): + return connect.NewError(connect.CodeFailedPrecondition, customer.ErrHasLiveRecords) case errors.Is(err, customer.ErrNotFound): return connect.NewError(connect.CodeNotFound, ErrCustomerNotFound) case errors.Is(err, checkout.ErrAlreadySubscribed): diff --git a/internal/api/v1beta1connect/billing_errors_test.go b/internal/api/v1beta1connect/billing_errors_test.go index 201933a029..e3e3706286 100644 --- a/internal/api/v1beta1connect/billing_errors_test.go +++ b/internal/api/v1beta1connect/billing_errors_test.go @@ -107,6 +107,12 @@ func TestMapBillingError(t *testing.T) { wantCode: connect.CodeFailedPrecondition, wantMsg: customer.ErrExistingAccountWithPendingDues.Error(), }, + { + name: "live records block the account delete", + err: fmt.Errorf("DeleteBillingAccount.Delete: customer_id=x: %w", customer.ErrHasLiveRecords), + wantCode: connect.CodeFailedPrecondition, + wantMsg: customer.ErrHasLiveRecords.Error(), + }, { name: "customer not found", err: fmt.Errorf("CreateCheckout.GetBillingAccountFromOrgID: org_id=abc: %w", customer.ErrNotFound), diff --git a/internal/store/postgres/billing_checkout_repository.go b/internal/store/postgres/billing_checkout_repository.go index 5d5cf481d8..f0a6d8ccf7 100644 --- a/internal/store/postgres/billing_checkout_repository.go +++ b/internal/store/postgres/billing_checkout_repository.go @@ -282,7 +282,7 @@ func (r BillingCheckoutRepository) UpdateByID(ctx context.Context, toUpdate chec } func (r BillingCheckoutRepository) DeleteByCustomerID(ctx context.Context, customerID string) error { - query, params, err := dialect.Delete(TABLE_BILLING_CHECKOUTS).Where(goqu.Ex{ + query, params, err := softDelete(TABLE_BILLING_CHECKOUTS).Where(goqu.Ex{ "customer_id": customerID, }).ToSQL() if err != nil { diff --git a/internal/store/postgres/billing_checkout_repository_pg_test.go b/internal/store/postgres/billing_checkout_repository_pg_test.go index 9d54f3d2d0..1b8be120d5 100644 --- a/internal/store/postgres/billing_checkout_repository_pg_test.go +++ b/internal/store/postgres/billing_checkout_repository_pg_test.go @@ -99,3 +99,21 @@ func (s *BillingCheckoutRepositoryPGTestSuite) TestListSkipsDeleted() { func TestBillingCheckoutRepositoryPG(t *testing.T) { suite.Run(t, new(BillingCheckoutRepositoryPGTestSuite)) } + +func (s *BillingCheckoutRepositoryPGTestSuite) TestDeleteByCustomerIDKeepsTheRowsAndHidesThem() { + before := scalarSQL(s.T(), s.ctx, s.client, + `SELECT deleted_at::text FROM billing_checkouts WHERE provider_id = 'bch-gone-session'`) + + s.Require().NoError(s.repository.DeleteByCustomerID(s.ctx, s.customerID())) + + s.Equal("2", scalarSQL(s.T(), s.ctx, s.client, + `SELECT count(*)::text FROM billing_checkouts WHERE customer_id = $1 AND deleted_at IS NOT NULL`, s.customerID())) + s.Equal(before, scalarSQL(s.T(), s.ctx, s.client, + `SELECT deleted_at::text FROM billing_checkouts WHERE provider_id = 'bch-gone-session'`), + "a row deleted earlier keeps its first delete time") + got, err := s.repository.List(s.ctx, checkout.Filter{CustomerID: s.customerID()}) + s.Require().NoError(err) + s.Empty(got) + _, err = s.repository.GetByID(s.ctx, s.checkoutID("bch-live-session")) + s.ErrorIs(err, checkout.ErrNotFound) +} diff --git a/internal/store/postgres/billing_customer_repository.go b/internal/store/postgres/billing_customer_repository.go index cfe9a1ac01..4d75ac9524 100644 --- a/internal/store/postgres/billing_customer_repository.go +++ b/internal/store/postgres/billing_customer_repository.go @@ -14,6 +14,7 @@ import ( "github.com/raystack/frontier/pkg/utils" "github.com/doug-martin/goqu/v9" + "github.com/doug-martin/goqu/v9/exp" "github.com/jmoiron/sqlx/types" "github.com/raystack/frontier/billing/customer" "github.com/raystack/frontier/pkg/auditrecord" @@ -497,16 +498,40 @@ func (r BillingCustomerRepository) UpdateDetailsByID(ctx context.Context, custom } func (r BillingCustomerRepository) Delete(ctx context.Context, id string) error { - query, params, err := dialect.Delete(TABLE_BILLING_CUSTOMERS). + lockQuery, lockParams, err := fromLive(TABLE_BILLING_CUSTOMERS). + Select("id"). + Where(goqu.Ex{"id": id}). + ForUpdate(exp.Wait). + ToSQL() + if err != nil { + return fmt.Errorf("%w: %w", errParse, err) + } + query, params, err := softDelete(TABLE_BILLING_CUSTOMERS). Where(goqu.Ex{"id": id}). Returning(&Customer{}).ToSQL() if err != nil { return fmt.Errorf("%w: %w", errParse, err) } + liveRecordsQuery, liveRecordsParams, err := liveBillingRecordsQuery(id) + if err != nil { + return fmt.Errorf("%w: %w", errParse, err) + } var customerModel Customer if err = r.dbc.WithTxn(ctx, sql.TxOptions{}, func(tx *sqlx.Tx) error { return r.dbc.WithTimeout(ctx, TABLE_BILLING_CUSTOMERS, "Delete", func(ctx context.Context) error { + var lockedID string + if err := tx.QueryRowxContext(ctx, lockQuery, lockParams...).Scan(&lockedID); err != nil { + return err + } + + var liveRecords bool + if err := tx.QueryRowxContext(ctx, liveRecordsQuery, liveRecordsParams...).Scan(&liveRecords); err != nil { + return err + } + if liveRecords { + return customer.ErrHasLiveRecords + } if err := tx.QueryRowxContext(ctx, query, params...).StructScan(&customerModel); err != nil { if errors.Is(err, sql.ErrNoRows) { return customer.ErrNotFound @@ -542,8 +567,10 @@ func (r BillingCustomerRepository) Delete(ctx context.Context, id string) error }); err != nil { err = checkPostgresError(err) switch { - case errors.Is(err, sql.ErrNoRows): + case errors.Is(err, customer.ErrNotFound), errors.Is(err, sql.ErrNoRows): return customer.ErrNotFound + case errors.Is(err, customer.ErrHasLiveRecords): + return customer.ErrHasLiveRecords default: return fmt.Errorf("%w: %w", errTxn, err) } @@ -551,3 +578,11 @@ func (r BillingCustomerRepository) Delete(ctx context.Context, id string) error return nil } + +func liveBillingRecordsQuery(customerID string) (string, []any, error) { + return dialect.Select(goqu.Or( + goqu.L("EXISTS (SELECT 1 FROM billing_subscriptions WHERE customer_id = ? AND deleted_at IS NULL)", customerID), + goqu.L("EXISTS (SELECT 1 FROM billing_invoices WHERE customer_id = ? AND deleted_at IS NULL)", customerID), + goqu.L("EXISTS (SELECT 1 FROM billing_checkouts WHERE customer_id = ? AND deleted_at IS NULL)", customerID), + )).ToSQL() +} diff --git a/internal/store/postgres/billing_customer_repository_pg_test.go b/internal/store/postgres/billing_customer_repository_pg_test.go index c0f24b2b9a..5546f4c04b 100644 --- a/internal/store/postgres/billing_customer_repository_pg_test.go +++ b/internal/store/postgres/billing_customer_repository_pg_test.go @@ -4,9 +4,15 @@ import ( "context" "fmt" "testing" + "time" + "github.com/raystack/frontier/billing/checkout" + "github.com/raystack/frontier/billing/credit" "github.com/raystack/frontier/billing/customer" + "github.com/raystack/frontier/billing/invoice" + "github.com/raystack/frontier/billing/subscription" "github.com/raystack/frontier/internal/store/postgres" + "github.com/raystack/frontier/pkg/auditrecord" "github.com/raystack/frontier/pkg/db" "github.com/stretchr/testify/suite" ) @@ -18,6 +24,11 @@ type BillingCustomerRepositoryPGTestSuite struct { ctx context.Context client *db.Client repository *postgres.BillingCustomerRepository + + subscriptions *postgres.BillingSubscriptionRepository + invoices *postgres.BillingInvoiceRepository + checkouts *postgres.BillingCheckoutRepository + transactions *postgres.BillingTransactionRepository } func (s *BillingCustomerRepositoryPGTestSuite) SetupSuite() { @@ -28,6 +39,10 @@ func (s *BillingCustomerRepositoryPGTestSuite) SetupSuite() { } s.ctx = context.TODO() s.repository = postgres.NewBillingCustomerRepository(s.client) + s.subscriptions = postgres.NewBillingSubscriptionRepository(s.client) + s.invoices = postgres.NewBillingInvoiceRepository(s.client) + s.checkouts = postgres.NewBillingCheckoutRepository(s.client) + s.transactions = postgres.NewBillingTransactionRepository(s.client) } func (s *BillingCustomerRepositoryPGTestSuite) TearDownSuite() { @@ -47,7 +62,7 @@ func (s *BillingCustomerRepositoryPGTestSuite) SetupTest() { func (s *BillingCustomerRepositoryPGTestSuite) TearDownTest() { queries := []string{} - for _, table := range []string{postgres.TABLE_BILLING_CUSTOMERS, postgres.TABLE_ORGANIZATIONS} { + for _, table := range []string{postgres.TABLE_BILLING_TRANSACTIONS, postgres.TABLE_BILLING_CHECKOUTS, postgres.TABLE_BILLING_INVOICES, postgres.TABLE_BILLING_SUBSCRIPTIONS, postgres.TABLE_BILLING_PLANS, postgres.TABLE_BILLING_CUSTOMERS, postgres.TABLE_ORGANIZATIONS, postgres.TABLE_AUDITRECORDS} { queries = append(queries, fmt.Sprintf("TRUNCATE TABLE %s RESTART IDENTITY CASCADE", table)) } if err := execQueries(s.ctx, s.client, queries); err != nil { @@ -119,3 +134,161 @@ func (s *BillingCustomerRepositoryPGTestSuite) TestUpdateCreditMinByIDSkipsDelet func TestBillingCustomerRepositoryPG(t *testing.T) { suite.Run(t, new(BillingCustomerRepositoryPGTestSuite)) } + +func (s *BillingCustomerRepositoryPGTestSuite) TestDeleteKeepsTheRowAndHidesIt() { + id := s.customerID("bc-cust-live") + + s.Require().NoError(s.repository.Delete(s.ctx, id)) + + deletedAt := scalarSQL(s.T(), s.ctx, s.client, + `SELECT coalesce(deleted_at::text, '') FROM billing_customers WHERE id = $1`, id) + s.NotEmpty(deletedAt, "the row stays with its delete time") + occurredAt := scalarSQL(s.T(), s.ctx, s.client, + `SELECT occurred_at::text FROM audit_records WHERE event = $1 AND resource_id = $2`, + string(auditrecord.BillingCustomerDeletedEvent), id) + s.Equal(deletedAt, occurredAt, "the audit record carries the delete time") + _, err := s.repository.GetByID(s.ctx, id) + s.ErrorIs(err, customer.ErrNotFound) + s.ErrorIs(s.repository.Delete(s.ctx, id), customer.ErrNotFound, "a second delete answers not found") +} + +func (s *BillingCustomerRepositoryPGTestSuite) TestDeleteRefusesWhileLiveRecordsPointAtTheAccount() { + cases := []struct { + table string + seed string + }{ + { + table: "billing_subscriptions", + seed: `INSERT INTO billing_subscriptions (customer_id, provider_id, plan_id, state) + VALUES ($1, 'bc-sub', (SELECT id FROM billing_plans WHERE name = 'bc-plan'), 'active')`, + }, + { + table: "billing_invoices", + seed: `INSERT INTO billing_invoices (customer_id, provider_id, amount, currency, hosted_url, state) + VALUES ($1, 'bc-inv', 100, 'usd', 'bc-inv', 'open')`, + }, + { + table: "billing_checkouts", + seed: `INSERT INTO billing_checkouts (customer_id, provider_id, checkout_url, state) + VALUES ($1, 'bc-session', 'bc-session', 'pending')`, + }, + } + for _, tc := range cases { + s.Run(tc.table, func() { + s.exec(`INSERT INTO billing_plans (name, description) VALUES ('bc-plan', 'plan under test')`) + id := s.customerID("bc-cust-live") + s.exec(tc.seed, id) + + s.ErrorIs(s.repository.Delete(s.ctx, id), customer.ErrHasLiveRecords) + s.Equal("", scalarSQL(s.T(), s.ctx, s.client, + `SELECT coalesce(deleted_at::text, '') FROM billing_customers WHERE id = $1`, id), + "the refused delete leaves the row live") + + s.exec(fmt.Sprintf(`UPDATE %s SET deleted_at = now() WHERE customer_id = $1`, tc.table), id) + s.NoError(s.repository.Delete(s.ctx, id), "a deleted child no longer blocks") + + s.exec(`TRUNCATE TABLE billing_plans RESTART IDENTITY CASCADE`) + s.exec(`UPDATE billing_customers SET deleted_at = NULL WHERE id = $1`, id) + s.exec(fmt.Sprintf(`DELETE FROM %s`, tc.table)) + }) + } +} + +func (s *BillingCustomerRepositoryPGTestSuite) seedChildRows(cust string) (subID, invID string) { + s.T().Helper() + s.exec(`INSERT INTO billing_plans (name, description) VALUES ('bc-plan', 'plan under test')`) + s.exec(`INSERT INTO billing_subscriptions (customer_id, provider_id, plan_id, state) + VALUES ($1, 'bc-sub', (SELECT id FROM billing_plans WHERE name = 'bc-plan'), 'canceled')`, cust) + s.exec(`INSERT INTO billing_invoices (customer_id, provider_id, amount, currency, hosted_url, state) + VALUES ($1, 'bc-inv', 100, 'usd', 'bc-inv', 'paid')`, cust) + s.exec(`INSERT INTO billing_checkouts (customer_id, provider_id, checkout_url, state) + VALUES ($1, 'bc-session', 'bc-session', 'expired')`, cust) + s.exec(`INSERT INTO billing_transactions (account_id, type, source, amount, description) + VALUES ($1, 'credit', 'system.buy', 100, 'bought'), ($1, 'debit', 'usage', 30, 'spent')`, cust) + subID = scalarSQL(s.T(), s.ctx, s.client, `SELECT id FROM billing_subscriptions WHERE provider_id = 'bc-sub'`) + invID = scalarSQL(s.T(), s.ctx, s.client, `SELECT id FROM billing_invoices WHERE provider_id = 'bc-inv'`) + return subID, invID +} + +func (s *BillingCustomerRepositoryPGTestSuite) deletedRows(table, column, id string) string { + s.T().Helper() + return scalarSQL(s.T(), s.ctx, s.client, + fmt.Sprintf(`SELECT count(*)::text FROM %s WHERE %s = $1 AND deleted_at IS NOT NULL`, table, column), id) +} + +func (s *BillingCustomerRepositoryPGTestSuite) TestTeardownInTheDeleterOrderKeepsEveryRow() { + cust := s.customerID("bc-cust-live") + subID, invID := s.seedChildRows(cust) + + s.Require().NoError(s.subscriptions.Delete(s.ctx, subID)) + s.Require().NoError(s.invoices.Delete(s.ctx, invID)) + s.Require().NoError(s.checkouts.DeleteByCustomerID(s.ctx, cust)) + s.Require().NoError(s.transactions.DeleteByAccountID(s.ctx, cust)) + s.Require().NoError(s.repository.Delete(s.ctx, cust)) + + s.Equal("1", s.deletedRows(postgres.TABLE_BILLING_SUBSCRIPTIONS, "customer_id", cust)) + s.Equal("1", s.deletedRows(postgres.TABLE_BILLING_INVOICES, "customer_id", cust)) + s.Equal("1", s.deletedRows(postgres.TABLE_BILLING_CHECKOUTS, "customer_id", cust)) + s.Equal("2", s.deletedRows(postgres.TABLE_BILLING_TRANSACTIONS, "account_id", cust)) + s.Equal("1", s.deletedRows(postgres.TABLE_BILLING_CUSTOMERS, "id", cust)) + + subs, err := s.subscriptions.List(s.ctx, subscription.Filter{CustomerID: cust}) + s.Require().NoError(err) + s.Empty(subs) + invs, err := s.invoices.List(s.ctx, invoice.Filter{CustomerID: cust}) + s.Require().NoError(err) + s.Empty(invs) + chks, err := s.checkouts.List(s.ctx, checkout.Filter{CustomerID: cust}) + s.Require().NoError(err) + s.Empty(chks) + balance, err := s.transactions.GetBalance(s.ctx, cust) + s.Require().NoError(err) + s.Equal(int64(0), balance) + custs, err := s.repository.List(s.ctx, customer.Filter{OrgID: s.orgID("bc-live")}) + s.Require().NoError(err) + s.Empty(custs) + txns, err := s.transactions.List(s.ctx, credit.Filter{CustomerID: cust}) + s.Require().NoError(err) + s.Empty(txns) +} + +func (s *BillingCustomerRepositoryPGTestSuite) TestTeardownCanBeRunAgain() { + cust := s.customerID("bc-cust-live") + subID, invID := s.seedChildRows(cust) + s.Require().NoError(s.subscriptions.Delete(s.ctx, subID)) + s.Require().NoError(s.invoices.Delete(s.ctx, invID)) + s.Require().NoError(s.checkouts.DeleteByCustomerID(s.ctx, cust)) + + s.Require().NoError(s.subscriptions.Delete(s.ctx, subID)) + s.Require().NoError(s.invoices.Delete(s.ctx, invID)) + s.Require().NoError(s.checkouts.DeleteByCustomerID(s.ctx, cust)) + s.Require().NoError(s.transactions.DeleteByAccountID(s.ctx, cust)) + s.Require().NoError(s.transactions.DeleteByAccountID(s.ctx, cust)) + s.Require().NoError(s.repository.Delete(s.ctx, cust)) + s.ErrorIs(s.repository.Delete(s.ctx, cust), customer.ErrNotFound) +} + +func (s *BillingCustomerRepositoryPGTestSuite) TestDeleteWaitsForAnInFlightChildInsertAndRefuses() { + id := s.customerID("bc-cust-live") + tx, err := s.client.DB.BeginTxx(s.ctx, nil) + s.Require().NoError(err) + defer tx.Rollback() //nolint:errcheck + var holderPID int + s.Require().NoError(tx.QueryRowContext(s.ctx, "SELECT pg_backend_pid()").Scan(&holderPID)) + _, err = tx.ExecContext(s.ctx, `INSERT INTO billing_checkouts (customer_id, provider_id, checkout_url, state) + VALUES ($1, 'bc-session', 'bc-session', 'pending')`, id) + s.Require().NoError(err) + + result := make(chan error, 1) + go func() { result <- s.repository.Delete(s.ctx, id) }() + + s.Require().Eventually(func() bool { + waiting, err := blockedBy(s.ctx, s.client, holderPID) + return err == nil && waiting == 1 + }, 5*time.Second, 20*time.Millisecond, "the delete waits on the row lock") + + s.Require().NoError(tx.Commit()) + s.ErrorIs(<-result, customer.ErrHasLiveRecords) + s.Equal("", scalarSQL(s.T(), s.ctx, s.client, + `SELECT coalesce(deleted_at::text, '') FROM billing_customers WHERE id = $1`, id)) +} diff --git a/internal/store/postgres/billing_invoice_repository.go b/internal/store/postgres/billing_invoice_repository.go index 61649e417a..d6ae85d5db 100644 --- a/internal/store/postgres/billing_invoice_repository.go +++ b/internal/store/postgres/billing_invoice_repository.go @@ -333,7 +333,7 @@ func (r BillingInvoiceRepository) UpdateByID(ctx context.Context, toUpdate invoi } func (r BillingInvoiceRepository) Delete(ctx context.Context, id string) error { - query, params, err := dialect.Delete(TABLE_BILLING_INVOICES).Where(goqu.Ex{ + query, params, err := softDelete(TABLE_BILLING_INVOICES).Where(goqu.Ex{ "id": id, }).ToSQL() if err != nil { diff --git a/internal/store/postgres/billing_invoice_repository_pg_test.go b/internal/store/postgres/billing_invoice_repository_pg_test.go index 4d54b1f83f..043411b9e4 100644 --- a/internal/store/postgres/billing_invoice_repository_pg_test.go +++ b/internal/store/postgres/billing_invoice_repository_pg_test.go @@ -187,3 +187,18 @@ func (s *BillingInvoiceRepositoryPGTestSuite) TestProviderIDIsUniqueOverLiveRows func TestBillingInvoiceRepositoryPG(t *testing.T) { suite.Run(t, new(BillingInvoiceRepositoryPGTestSuite)) } + +func (s *BillingInvoiceRepositoryPGTestSuite) TestDeleteKeepsTheRowAndHidesIt() { + id := s.invoiceID("bi-inv-live") + + s.Require().NoError(s.repository.Delete(s.ctx, id)) + + deletedAt := scalarSQL(s.T(), s.ctx, s.client, + `SELECT coalesce(deleted_at::text, '') FROM billing_invoices WHERE id = $1`, id) + s.NotEmpty(deletedAt) + _, err := s.repository.GetByID(s.ctx, id) + s.ErrorIs(err, invoice.ErrNotFound) + s.Require().NoError(s.repository.Delete(s.ctx, id)) + s.Equal(deletedAt, scalarSQL(s.T(), s.ctx, s.client, + `SELECT deleted_at::text FROM billing_invoices WHERE id = $1`, id), "a second delete keeps the first delete time") +} diff --git a/internal/store/postgres/billing_subscription_repository.go b/internal/store/postgres/billing_subscription_repository.go index 4396d7ed59..ae77018d6c 100644 --- a/internal/store/postgres/billing_subscription_repository.go +++ b/internal/store/postgres/billing_subscription_repository.go @@ -450,7 +450,7 @@ func (r BillingSubscriptionRepository) List(ctx context.Context, filter subscrip } func (r BillingSubscriptionRepository) Delete(ctx context.Context, id string) error { - stmt := dialect.Delete(TABLE_BILLING_SUBSCRIPTIONS).Where(goqu.Ex{ + stmt := softDelete(TABLE_BILLING_SUBSCRIPTIONS).Where(goqu.Ex{ "id": id, }) query, params, err := stmt.ToSQL() diff --git a/internal/store/postgres/billing_subscription_repository_pg_test.go b/internal/store/postgres/billing_subscription_repository_pg_test.go index 5e125915d0..91fc72244d 100644 --- a/internal/store/postgres/billing_subscription_repository_pg_test.go +++ b/internal/store/postgres/billing_subscription_repository_pg_test.go @@ -126,3 +126,21 @@ func (s *BillingSubscriptionRepositoryPGTestSuite) TestProviderIDIsUniqueOverLiv func TestBillingSubscriptionRepositoryPG(t *testing.T) { suite.Run(t, new(BillingSubscriptionRepositoryPGTestSuite)) } + +func (s *BillingSubscriptionRepositoryPGTestSuite) TestDeleteKeepsTheRowAndHidesIt() { + id := s.subscriptionID("bs-sub-live") + + s.Require().NoError(s.repository.Delete(s.ctx, id)) + + deletedAt := scalarSQL(s.T(), s.ctx, s.client, + `SELECT coalesce(deleted_at::text, '') FROM billing_subscriptions WHERE id = $1`, id) + s.NotEmpty(deletedAt) + _, err := s.repository.GetByID(s.ctx, id) + s.ErrorIs(err, subscription.ErrNotFound) + got, err := s.repository.List(s.ctx, subscription.Filter{CustomerID: s.customerID()}) + s.Require().NoError(err) + s.Empty(got) + s.Require().NoError(s.repository.Delete(s.ctx, id)) + s.Equal(deletedAt, scalarSQL(s.T(), s.ctx, s.client, + `SELECT deleted_at::text FROM billing_subscriptions WHERE id = $1`, id), "a second delete keeps the first delete time") +} diff --git a/internal/store/postgres/billing_transactions_repository.go b/internal/store/postgres/billing_transactions_repository.go index 668b450cd2..10f98932ed 100644 --- a/internal/store/postgres/billing_transactions_repository.go +++ b/internal/store/postgres/billing_transactions_repository.go @@ -143,6 +143,12 @@ func (r BillingTransactionRepository) CreateEntry(ctx context.Context, debitEntr } } + if creditEntry.CustomerID != schema.PlatformOrgID.String() { + if _, err = r.customerRepo.GetByID(ctx, creditEntry.CustomerID); err != nil { + return nil, fmt.Errorf("failed to get customer account: %w", err) + } + } + var creditReturnedEntry, debitReturnedEntry credit.Transaction err = r.withRetry(ctx, func() error { return r.dbc.WithTxn(ctx, txOpts, func(tx *sqlx.Tx) error { @@ -307,7 +313,7 @@ func (r BillingTransactionRepository) UpdateByID(ctx context.Context, toUpdate c } func (r BillingTransactionRepository) DeleteByAccountID(ctx context.Context, accountID string) error { - query, params, err := dialect.Delete(TABLE_BILLING_TRANSACTIONS).Where(goqu.Ex{ + query, params, err := softDelete(TABLE_BILLING_TRANSACTIONS).Where(goqu.Ex{ "account_id": accountID, }).ToSQL() if err != nil { diff --git a/internal/store/postgres/billing_transactions_repository_pg_test.go b/internal/store/postgres/billing_transactions_repository_pg_test.go index fb14b8f80d..ab78d9546f 100644 --- a/internal/store/postgres/billing_transactions_repository_pg_test.go +++ b/internal/store/postgres/billing_transactions_repository_pg_test.go @@ -63,7 +63,7 @@ func (s *BillingTransactionRepositoryTestSuite) TestCreateEntry() { } creditEntry := credit.Transaction{ ID: uuid.New().String(), - CustomerID: uuid.New().String(), + CustomerID: s.customer("bt-credit-target", 0), Type: credit.CreditType, Amount: 10, Source: credit.SourceSystemAwardedEvent, @@ -191,6 +191,39 @@ func (s *BillingTransactionRepositoryTestSuite) TestCreateEntryRefusesDeletedCus s.ErrorIs(s.spend(account, 10), customer.ErrNotFound) } +func (s *BillingTransactionRepositoryTestSuite) TestCreateEntryRefusesACreditToADeletedCustomer() { + account := s.customer("bt-credit-gone", 0) + s.exec(`UPDATE billing_customers SET deleted_at = now() WHERE id = $1`, account) + + _, err := s.repository.CreateEntry(s.ctx, + credit.Transaction{ID: uuid.New().String(), CustomerID: schema.PlatformOrgID.String(), Type: credit.DebitType, Amount: 10, Source: credit.SourceSystemAwardedEvent}, + credit.Transaction{CustomerID: account, Type: credit.CreditType, Amount: 10, Source: credit.SourceSystemAwardedEvent}) + + s.ErrorIs(err, customer.ErrNotFound) + s.Equal("0", scalarSQL(s.T(), s.ctx, s.client, `SELECT count(*)::text FROM billing_transactions WHERE account_id = $1`, account)) +} + func TestBillingTransactionRepository(t *testing.T) { suite.Run(t, new(BillingTransactionRepositoryTestSuite)) } + +func (s *BillingTransactionRepositoryTestSuite) TestDeleteByAccountIDKeepsTheRowsAndZeroesTheBalance() { + account := uuid.New().String() + s.txn(account, string(credit.CreditType), credit.SourceSystemBuyEvent, 100) + s.txn(account, string(credit.DebitType), "usage", 30) + gone := s.txn(account, string(credit.CreditType), credit.SourceSystemAwardedEvent, 5) + s.softDelete(gone) + before := scalarSQL(s.T(), s.ctx, s.client, `SELECT deleted_at::text FROM billing_transactions WHERE id = $1`, gone) + + s.Require().NoError(s.repository.DeleteByAccountID(s.ctx, account)) + + s.Equal("3", scalarSQL(s.T(), s.ctx, s.client, + `SELECT count(*)::text FROM billing_transactions WHERE account_id = $1 AND deleted_at IS NOT NULL`, account)) + s.Equal(before, scalarSQL(s.T(), s.ctx, s.client, `SELECT deleted_at::text FROM billing_transactions WHERE id = $1`, gone)) + balance, err := s.repository.GetBalance(s.ctx, account) + s.Require().NoError(err) + s.Equal(int64(0), balance) + got, err := s.repository.List(s.ctx, credit.Filter{CustomerID: account}) + s.Require().NoError(err) + s.Empty(got) +}