feat(store): billing invoices, transactions and cusomter reads to skip soft-deleted rows

core/deleter/service.go

@@ -407,6 +407,11 @@ func (d Service) deleteCustomers(ctx context.Context, id string, customers []cus
 		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)
 		}

internal/store/postgres/billing_customer_repository.go

@@ -192,7 +192,7 @@ func (r BillingCustomerRepository) Create(ctx context.Context, toCreate customer
 }
 
 func (r BillingCustomerRepository) GetByID(ctx context.Context, id string) (customer.Customer, error) {
-	stmt := dialect.Select().From(TABLE_BILLING_CUSTOMERS).Where(goqu.Ex{
+	stmt := fromLive(TABLE_BILLING_CUSTOMERS).Where(goqu.Ex{
 		"id": id,
 	})
 	query, params, err := stmt.ToSQL()
@@ -216,7 +216,7 @@ func (r BillingCustomerRepository) GetByID(ctx context.Context, id string) (cust
 }
 
 func (r BillingCustomerRepository) List(ctx context.Context, flt customer.Filter) ([]customer.Customer, error) {
-	stmt := dialect.Select().From(TABLE_BILLING_CUSTOMERS).Order(goqu.I("created_at").Desc())
+	stmt := fromLive(TABLE_BILLING_CUSTOMERS).Order(goqu.I("created_at").Desc())
 
 	if flt.OrgID != "" {
 		stmt = stmt.Where(goqu.Ex{
@@ -407,7 +407,7 @@ func (r BillingCustomerRepository) UpdateCreditMinByID(ctx context.Context, cust
 }
 
 func (r BillingCustomerRepository) GetDetailsByID(ctx context.Context, customerID string) (customer.Details, error) {
-	stmt := dialect.Select("credit_min", "due_in_days").From(TABLE_BILLING_CUSTOMERS).Where(goqu.Ex{
+	stmt := fromLive(TABLE_BILLING_CUSTOMERS).Select("credit_min", "due_in_days").Where(goqu.Ex{
 		"id": customerID,
 	})
 	query, params, err := stmt.ToSQL()

internal/store/postgres/billing_customer_repository_pg_test.go

@@ -0,0 +1,106 @@
+package postgres_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/raystack/frontier/billing/customer"
+	"github.com/raystack/frontier/internal/store/postgres"
+	"github.com/raystack/frontier/pkg/db"
+	"github.com/stretchr/testify/suite"
+)
+
+// Runs the billing customer reads against a real postgres to check that a
+// soft-deleted customer stays out of every read.
+type BillingCustomerRepositoryPGTestSuite struct {
+	suite.Suite
+	ctx        context.Context
+	client     *db.Client
+	repository *postgres.BillingCustomerRepository
+}
+
+func (s *BillingCustomerRepositoryPGTestSuite) SetupSuite() {
+	var err error
+	s.client, err = newTestClient()
+	if err != nil {
+		s.T().Fatal(err)
+	}
+	s.ctx = context.TODO()
+	s.repository = postgres.NewBillingCustomerRepository(s.client)
+}
+
+func (s *BillingCustomerRepositoryPGTestSuite) TearDownSuite() {
+	if err := closeTestClient(s.client); err != nil {
+		s.T().Fatal(err)
+	}
+}
+
+func (s *BillingCustomerRepositoryPGTestSuite) SetupTest() {
+	s.exec(`INSERT INTO organizations (name, title) VALUES ('bc-live', 'Live Org')`)
+
+	s.customer("bc-cust-live", "bc-live", -100)
+	s.customer("bc-cust-gone", "bc-live", -100)
+
+	s.exec(`UPDATE billing_customers SET deleted_at = now() WHERE name = 'bc-cust-gone'`)
+}
+
+func (s *BillingCustomerRepositoryPGTestSuite) TearDownTest() {
+	queries := []string{}
+	for _, table := range []string{postgres.TABLE_BILLING_CUSTOMERS, postgres.TABLE_ORGANIZATIONS} {
+		queries = append(queries, fmt.Sprintf("TRUNCATE TABLE %s RESTART IDENTITY CASCADE", table))
+	}
+	if err := execQueries(s.ctx, s.client, queries); err != nil {
+		s.T().Fatal(err)
+	}
+}
+
+func (s *BillingCustomerRepositoryPGTestSuite) exec(query string, args ...any) {
+	s.T().Helper()
+	execSQL(s.T(), s.ctx, s.client, query, args...)
+}
+
+func (s *BillingCustomerRepositoryPGTestSuite) orgID(name string) string {
+	s.T().Helper()
+	return scalarSQL(s.T(), s.ctx, s.client, `SELECT id FROM organizations WHERE name = $1`, name)
+}
+
+func (s *BillingCustomerRepositoryPGTestSuite) customerID(name string) string {
+	s.T().Helper()
+	return scalarSQL(s.T(), s.ctx, s.client, `SELECT id FROM billing_customers WHERE name = $1`, name)
+}
+
+func (s *BillingCustomerRepositoryPGTestSuite) customer(name, orgName string, creditMin int64) {
+	s.T().Helper()
+	s.exec(`INSERT INTO billing_customers (org_id, provider_id, name, email, credit_min)
+		VALUES ($1, $2, $2, $2, $3)`, s.orgID(orgName), name, creditMin)
+}
+
+func (s *BillingCustomerRepositoryPGTestSuite) TestGetByIDSkipsDeleted() {
+	got, err := s.repository.GetByID(s.ctx, s.customerID("bc-cust-live"))
+	s.Require().NoError(err)
+	s.Equal("bc-cust-live", got.Name)
+
+	_, err = s.repository.GetByID(s.ctx, s.customerID("bc-cust-gone"))
+	s.ErrorIs(err, customer.ErrNotFound)
+}
+
+func (s *BillingCustomerRepositoryPGTestSuite) TestListSkipsDeleted() {
+	got, err := s.repository.List(s.ctx, customer.Filter{OrgID: s.orgID("bc-live")})
+	s.Require().NoError(err)
+	s.Require().Len(got, 1)
+	s.Equal("bc-cust-live", got[0].Name)
+}
+
+func (s *BillingCustomerRepositoryPGTestSuite) TestGetDetailsByIDSkipsDeleted() {
+	got, err := s.repository.GetDetailsByID(s.ctx, s.customerID("bc-cust-live"))
+	s.Require().NoError(err)
+	s.Equal(int64(-100), got.CreditMin)
+
+	_, err = s.repository.GetDetailsByID(s.ctx, s.customerID("bc-cust-gone"))
+	s.ErrorIs(err, customer.ErrNotFound)
+}
+
+func TestBillingCustomerRepositoryPG(t *testing.T) {
+	suite.Run(t, new(BillingCustomerRepositoryPGTestSuite))
+}

internal/store/postgres/billing_invoice_repository.go

@@ -185,7 +185,7 @@ func (r BillingInvoiceRepository) Create(ctx context.Context, toCreate invoice.I
 }
 
 func (r BillingInvoiceRepository) GetByID(ctx context.Context, id string) (invoice.Invoice, error) {
-	stmt := dialect.Select().From(TABLE_BILLING_INVOICES).Where(goqu.Ex{
+	stmt := fromLive(TABLE_BILLING_INVOICES).Where(goqu.Ex{
 		"id": id,
 	})
 	query, params, err := stmt.ToSQL()
@@ -209,7 +209,7 @@ func (r BillingInvoiceRepository) GetByID(ctx context.Context, id string) (invoi
 }
 
 func (r BillingInvoiceRepository) List(ctx context.Context, flt invoice.Filter) ([]invoice.Invoice, error) {
-	stmt := dialect.Select().From(TABLE_BILLING_INVOICES)
+	stmt := fromLive(TABLE_BILLING_INVOICES)
 	if flt.CustomerID != "" {
 		stmt = stmt.Where(goqu.Ex{
 			"customer_id": flt.CustomerID,
@@ -396,7 +396,7 @@ func (r BillingInvoiceRepository) prepareDataQuery(rqlQuery *rql.Query) (string,
 }
 
 func (r BillingInvoiceRepository) buildBaseQuery() *goqu.SelectDataset {
-	return dialect.From(TABLE_BILLING_INVOICES).Prepared(true).
+	return fromLive(TABLE_BILLING_INVOICES).Prepared(true).
 		InnerJoin(
 			goqu.T(TABLE_BILLING_CUSTOMERS),
 			goqu.On(goqu.I(TABLE_BILLING_INVOICES+".customer_id").Eq(goqu.I(TABLE_BILLING_CUSTOMERS+".id"))),
@@ -405,6 +405,7 @@ func (r BillingInvoiceRepository) buildBaseQuery() *goqu.SelectDataset {
 			goqu.T(TABLE_ORGANIZATIONS),
 			goqu.On(goqu.I(TABLE_BILLING_CUSTOMERS+".org_id").Eq(goqu.I(TABLE_ORGANIZATIONS+".id"))),
 		).
+		Where(live(TABLE_BILLING_CUSTOMERS), live(TABLE_ORGANIZATIONS)).
 		Select(
 			goqu.I(TABLE_BILLING_INVOICES+".id").As("id"),
 			goqu.I(TABLE_BILLING_INVOICES+".amount").As("amount"),

internal/store/postgres/billing_invoice_repository_pg_test.go

@@ -0,0 +1,152 @@
+package postgres_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/raystack/frontier/billing/invoice"
+	"github.com/raystack/frontier/internal/store/postgres"
+	"github.com/raystack/frontier/pkg/db"
+	"github.com/raystack/frontier/pkg/pagination"
+	"github.com/raystack/salt/rql"
+	"github.com/stretchr/testify/suite"
+)
+
+// Runs the billing invoice reads against a real postgres to check that a
+// soft-deleted invoice stays out of every read, and that the admin search also
+// drops invoices whose billing customer or organization is soft-deleted.
+type BillingInvoiceRepositoryPGTestSuite struct {
+	suite.Suite
+	ctx        context.Context
+	client     *db.Client
+	repository *postgres.BillingInvoiceRepository
+}
+
+func (s *BillingInvoiceRepositoryPGTestSuite) SetupSuite() {
+	var err error
+	s.client, err = newTestClient()
+	if err != nil {
+		s.T().Fatal(err)
+	}
+	s.ctx = context.TODO()
+	s.repository = postgres.NewBillingInvoiceRepository(s.client)
+}
+
+func (s *BillingInvoiceRepositoryPGTestSuite) TearDownSuite() {
+	if err := closeTestClient(s.client); err != nil {
+		s.T().Fatal(err)
+	}
+}
+
+func (s *BillingInvoiceRepositoryPGTestSuite) SetupTest() {
+	s.exec(`INSERT INTO organizations (name, title) VALUES ('bi-live', 'Live Org'), ('bi-gone', 'Gone Org')`)
+
+	s.customer("bi-cust-live", "bi-live")
+	s.customer("bi-cust-gone", "bi-live")
+	s.customer("bi-cust-other", "bi-gone")
+
+	s.invoice("bi-cust-live", "bi-inv-live", "paid")
+	s.invoice("bi-cust-live", "bi-inv-deleted", "paid")
+	s.invoice("bi-cust-gone", "bi-inv-orphan", "open")
+	s.invoice("bi-cust-other", "bi-inv-otherorg", "paid")
+
+	s.exec(`UPDATE billing_invoices SET deleted_at = now() WHERE hosted_url = 'bi-inv-deleted'`)
+	s.exec(`UPDATE billing_customers SET deleted_at = now() WHERE name = 'bi-cust-gone'`)
+	s.exec(`UPDATE organizations SET deleted_at = now() WHERE name = 'bi-gone'`)
+}
+
+func (s *BillingInvoiceRepositoryPGTestSuite) TearDownTest() {
+	queries := []string{}
+	for _, table := range []string{postgres.TABLE_BILLING_INVOICES, postgres.TABLE_BILLING_CUSTOMERS,
+		postgres.TABLE_ORGANIZATIONS} {
+		queries = append(queries, fmt.Sprintf("TRUNCATE TABLE %s RESTART IDENTITY CASCADE", table))
+	}
+	if err := execQueries(s.ctx, s.client, queries); err != nil {
+		s.T().Fatal(err)
+	}
+}
+
+func (s *BillingInvoiceRepositoryPGTestSuite) exec(query string, args ...any) {
+	s.T().Helper()
+	execSQL(s.T(), s.ctx, s.client, query, args...)
+}
+
+func (s *BillingInvoiceRepositoryPGTestSuite) orgID(name string) string {
+	s.T().Helper()
+	return scalarSQL(s.T(), s.ctx, s.client, `SELECT id FROM organizations WHERE name = $1`, name)
+}
+
+func (s *BillingInvoiceRepositoryPGTestSuite) customerID(name string) string {
+	s.T().Helper()
+	return scalarSQL(s.T(), s.ctx, s.client, `SELECT id FROM billing_customers WHERE name = $1`, name)
+}
+
+func (s *BillingInvoiceRepositoryPGTestSuite) invoiceID(hostedURL string) string {
+	s.T().Helper()
+	return scalarSQL(s.T(), s.ctx, s.client, `SELECT id FROM billing_invoices WHERE hosted_url = $1`, hostedURL)
+}
+
+func (s *BillingInvoiceRepositoryPGTestSuite) customer(name, orgName string) {
+	s.T().Helper()
+	s.exec(`INSERT INTO billing_customers (org_id, provider_id, name, email)
+		VALUES ($1, $2, $2, $2)`, s.orgID(orgName), name)
+}
+
+func (s *BillingInvoiceRepositoryPGTestSuite) invoice(customer, hostedURL, state string) {
+	s.T().Helper()
+	s.exec(`INSERT INTO billing_invoices (customer_id, provider_id, amount, currency, hosted_url, state)
+		VALUES ((SELECT id FROM billing_customers WHERE name = $1), $2, 100, 'usd', $2, $3)`,
+		customer, hostedURL, state)
+}
+
+func (s *BillingInvoiceRepositoryPGTestSuite) links(invoices []invoice.Invoice) []string {
+	out := make([]string, 0, len(invoices))
+	for _, i := range invoices {
+		out = append(out, i.HostedURL)
+	}
+	return out
+}
+
+func (s *BillingInvoiceRepositoryPGTestSuite) TestGetByIDSkipsDeleted() {
+	got, err := s.repository.GetByID(s.ctx, s.invoiceID("bi-inv-live"))
+	s.Require().NoError(err)
+	s.Equal("bi-inv-live", got.HostedURL)
+
+	_, err = s.repository.GetByID(s.ctx, s.invoiceID("bi-inv-deleted"))
+	s.ErrorIs(err, invoice.ErrNotFound)
+}
+
+func (s *BillingInvoiceRepositoryPGTestSuite) TestListSkipsDeletedInvoicesAndCountsAgree() {
+	page := pagination.NewPagination(1, 10)
+	got, err := s.repository.List(s.ctx, invoice.Filter{
+		CustomerID: s.customerID("bi-cust-live"),
+		Pagination: page,
+	})
+	s.Require().NoError(err)
+	s.Equal([]string{"bi-inv-live"}, s.links(got))
+	s.Equal(int32(1), page.Count, "the count statement carries the same filter as the rows")
+}
+
+func (s *BillingInvoiceRepositoryPGTestSuite) TestListReadsOnlyTheInvoiceTable() {
+	// List has no join, so an invoice of a soft-deleted customer still lists.
+	// That matches every other single-table List in the store. The admin
+	// search below is where the customer and organization are checked.
+	got, err := s.repository.List(s.ctx, invoice.Filter{CustomerID: s.customerID("bi-cust-gone")})
+	s.Require().NoError(err)
+	s.Equal([]string{"bi-inv-orphan"}, s.links(got))
+}
+
+func (s *BillingInvoiceRepositoryPGTestSuite) TestSearchSkipsDeletedInvoicesCustomersAndOrgs() {
+	res, err := s.repository.Search(s.ctx, &rql.Query{Limit: 50})
+	s.Require().NoError(err)
+	out := make([]string, 0, len(res))
+	for _, i := range res {
+		out = append(out, i.InvoiceLink)
+	}
+	s.Equal([]string{"bi-inv-live"}, out)
+}
+
+func TestBillingInvoiceRepositoryPG(t *testing.T) {
+	suite.Run(t, new(BillingInvoiceRepositoryPGTestSuite))
+}