feat(policy): 1500 Attribute create with Values (one RPC Call) should employ a db transaction (#1778)

### Proposed Changes

fix #1500

- adds `PolicyDbClient.RunInTx()` method
- updates attribute create to use the method

### Checklist

- [ ] I have added or updated unit tests
- [ ] I have added or updated integration tests (if appropriate)
- [ ] I have added or updated documentation

### Testing Instructions

Author: ryanulit

service/integration/policy_test.go

@@ -0,0 +1,128 @@
+package integration
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"testing"
+
+	"github.com/opentdf/platform/service/internal/fixtures"
+	"github.com/opentdf/platform/service/policy/db"
+	"github.com/stretchr/testify/suite"
+)
+
+type PolicyDBClientSuite struct {
+	suite.Suite
+	f   fixtures.Fixtures
+	db  fixtures.DBInterface
+	ctx context.Context //nolint:containedctx // context is used in the test suite
+}
+
+func (s *PolicyDBClientSuite) SetupSuite() {
+	s.ctx = context.Background()
+	c := *Config
+	c.DB.Schema = "text_opentdf_policy_db_client"
+	s.db = fixtures.NewDBInterface(c)
+	s.f = fixtures.NewFixture(s.db)
+	s.f.Provision()
+}
+
+func (s *PolicyDBClientSuite) TearDownSuite() {
+	slog.Info("tearing down db.PolicyDbClient test suite")
+	s.f.TearDown()
+}
+
+func (s *PolicyDBClientSuite) Test_RunInTx_CommitsOnSuccess() {
+	var (
+		nsName    = "success.com"
+		attrName  = fmt.Sprintf("http://%s/attr/attr_one", nsName)
+		attrValue = fmt.Sprintf("http://%s/attr/%s/value/attr_one_value", nsName, attrName)
+
+		nsID   string
+		attrID string
+		valID  string
+		err    error
+	)
+
+	txErr := s.db.PolicyClient.RunInTx(s.ctx, func(txClient *db.PolicyDBClient) error {
+		nsID, err = txClient.Queries.CreateNamespace(s.ctx, db.CreateNamespaceParams{
+			Name: nsName,
+		})
+		s.Require().NoError(err)
+		s.Require().NotNil(nsID)
+
+		attrID, err = txClient.Queries.CreateAttribute(s.ctx, db.CreateAttributeParams{
+			NamespaceID: nsID,
+			Name:        attrName,
+			Rule:        db.AttributeDefinitionRuleALLOF,
+		})
+		s.Require().NoError(err)
+		s.Require().NotNil(attrID)
+
+		valID, err = txClient.Queries.CreateAttributeValue(s.ctx, db.CreateAttributeValueParams{
+			AttributeDefinitionID: attrID,
+			Value:                 attrValue,
+		})
+		s.Require().NoError(err)
+		s.Require().NotNil(valID)
+
+		return nil
+	})
+	s.Require().NoError(txErr)
+
+	ns, err := s.db.PolicyClient.GetNamespace(s.ctx, nsID)
+	s.Require().NoError(err)
+	s.Equal(nsName, ns.GetName())
+
+	attr, err := s.db.PolicyClient.GetAttribute(s.ctx, attrID)
+	s.Require().NoError(err)
+	s.Equal(attrName, attr.GetName())
+
+	attrVal, err := s.db.PolicyClient.GetAttributeValue(s.ctx, valID)
+	s.Require().NoError(err)
+	s.Equal(attrValue, attrVal.GetValue())
+}
+
+func (s *PolicyDBClientSuite) Test_RunInTx_RollsBackOnFailure() {
+	var (
+		nsName   = "failure.com"
+		attrName = fmt.Sprintf("http://%s/attr/attr_one", nsName)
+
+		nsID   string
+		attrID string
+		err    error
+	)
+
+	txErr := s.db.PolicyClient.RunInTx(s.ctx, func(txClient *db.PolicyDBClient) error {
+		nsID, err = txClient.Queries.CreateNamespace(s.ctx, db.CreateNamespaceParams{
+			Name: nsName,
+		})
+		s.Require().NoError(err)
+		s.Require().NotNil(nsID)
+
+		attrID, err = txClient.Queries.CreateAttribute(s.ctx, db.CreateAttributeParams{
+			NamespaceID: "invalid_ns_id",
+			Name:        attrName,
+			Rule:        db.AttributeDefinitionRuleALLOF,
+		})
+		s.Require().Error(err)
+		s.Require().Zero(attrID)
+		return err
+	})
+	s.Require().Error(txErr)
+
+	ns, err := s.db.PolicyClient.GetNamespace(s.ctx, nsID)
+	s.Require().Error(err)
+	s.Nil(ns)
+
+	attr, err := s.db.PolicyClient.GetAttribute(s.ctx, attrID)
+	s.Require().Error(err)
+	s.Nil(attr)
+}
+
+func TestPolicySuite(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping policy integration tests")
+	}
+	suite.Run(t, new(PolicyDBClientSuite))
+}

service/pkg/db/db.go

@@ -54,6 +54,7 @@ func (t Table) Field(field string) string {
 // We can rename this but wanted to get mocks working.
 type PgxIface interface {
 	Acquire(ctx context.Context) (*pgxpool.Conn, error)
+	Begin(ctx context.Context) (pgx.Tx, error)
 	Exec(context.Context, string, ...any) (pgconn.CommandTag, error)
 	QueryRow(context.Context, string, ...any) pgx.Row
 	Query(context.Context, string, ...any) (pgx.Rows, error)

service/pkg/db/errors.go

@@ -22,6 +22,9 @@ var (
 	ErrUUIDInvalid               = errors.New("ErrUUIDInvalid: value not a valid UUID")
 	ErrMissingValue              = errors.New("ErrMissingValue: value must be included")
 	ErrListLimitTooLarge         = errors.New("ErrListLimitTooLarge: requested limit greater than configured maximum")
+	ErrTxBeginFailed             = errors.New("ErrTxBeginFailed: failed to begin DB transaction")
+	ErrTxRollbackFailed          = errors.New("ErrTxRollbackFailed: failed to rollback DB transaction")
+	ErrTxCommitFailed            = errors.New("ErrTxCommitFailed: failed to commit DB transaction")
 )
 
 // Get helpful error message for PostgreSQL violation

service/policy/attributes/attributes.go

@@ -54,19 +54,26 @@ func (s AttributesService) CreateAttribute(ctx context.Context,
 		ActionType: audit.ActionTypeCreate,
 	}
 
-	item, err := s.dbClient.CreateAttribute(ctx, req.Msg)
+	err := s.dbClient.RunInTx(ctx, func(txClient *policydb.PolicyDBClient) error {
+		item, err := txClient.CreateAttribute(ctx, req.Msg)
+		if err != nil {
+			s.logger.Audit.PolicyCRUDFailure(ctx, auditParams)
+			return err
+		}
+
+		s.logger.Debug("created new attribute definition", slog.String("name", req.Msg.GetName()))
+
+		auditParams.ObjectID = item.GetId()
+		auditParams.Original = item
+		s.logger.Audit.PolicyCRUDSuccess(ctx, auditParams)
+
+		rsp.Attribute = item
+		return nil
+	})
 	if err != nil {
-		s.logger.Audit.PolicyCRUDFailure(ctx, auditParams)
 		return nil, db.StatusifyError(err, db.ErrTextCreationFailed, slog.String("attribute", req.Msg.String()))
 	}
 
-	s.logger.Debug("created new attribute definition", slog.String("name", req.Msg.GetName()))
-
-	auditParams.ObjectID = item.GetId()
-	auditParams.Original = item
-	s.logger.Audit.PolicyCRUDSuccess(ctx, auditParams)
-
-	rsp.Attribute = item
 	return connect.NewResponse(rsp), nil
 }
 

service/policy/db/policy.go

@@ -1,6 +1,9 @@
 package db
 
 import (
+	"context"
+	"fmt"
+
 	"github.com/opentdf/platform/protocol/go/common"
 	"github.com/opentdf/platform/service/logger"
 	"github.com/opentdf/platform/service/pkg/db"
@@ -31,6 +34,33 @@ func NewClient(c *db.Client, logger *logger.Logger, configuredListLimitMax, conf
 	return PolicyDBClient{c, logger, New(c.Pgx), ListConfig{limitDefault: configuredListLimitDefault, limitMax: configuredListLimitMax}}
 }
 
+func (c *PolicyDBClient) RunInTx(ctx context.Context, query func(txClient *PolicyDBClient) error) error {
+	tx, err := c.Client.Pgx.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("%w: %w", db.ErrTxBeginFailed, err)
+	}
+
+	txClient := &PolicyDBClient{c.Client, c.logger, c.Queries.WithTx(tx), c.listCfg}
+
+	err = query(txClient)
+	if err != nil {
+		c.logger.WarnContext(ctx, "error during DB transaction, rolling back")
+
+		if rollbackErr := tx.Rollback(ctx); rollbackErr != nil {
+			// this should never happen, but if it does, we want to know about it
+			return fmt.Errorf("%w, transaction [%w]: %w", db.ErrTxRollbackFailed, err, rollbackErr)
+		}
+
+		return err
+	}
+
+	if err = tx.Commit(ctx); err != nil {
+		return fmt.Errorf("%w: %w", db.ErrTxCommitFailed, err)
+	}
+
+	return nil
+}
+
 func getDBStateTypeTransformedEnum(state common.ActiveStateEnum) transformedState {
 	switch state.String() {
 	case common.ActiveStateEnum_ACTIVE_STATE_ENUM_ACTIVE.String():