feat(policy): Create/Update scs to use transaction. (#2882)

### Proposed Changes

1.) Add transactions around subject-condition-set creation/update. 

>[!NOTE]
>I believe this is the cause for all the intermittent/flaky otdfctl
>test failures that have been occurring.

### 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: c-r33d

service/policy/subjectmapping/subject_mapping.go

@@ -267,17 +267,27 @@ func (s SubjectMappingService) CreateSubjectConditionSet(ctx context.Context,
 		ObjectType: audit.ObjectTypeConditionSet,
 	}
 
-	conditionSet, err := s.dbClient.CreateSubjectConditionSet(ctx, req.Msg.GetSubjectConditionSet())
+	var conditionSet *policy.SubjectConditionSet
+	err := s.dbClient.RunInTx(ctx, func(txClient *policydb.PolicyDBClient) error {
+		cs, err := txClient.CreateSubjectConditionSet(ctx, req.Msg.GetSubjectConditionSet())
+		if err != nil {
+			s.logger.Audit.PolicyCRUDFailure(ctx, auditParams)
+			return db.StatusifyError(ctx, s.logger, err, db.ErrTextCreationFailed, slog.String("subjectConditionSet", req.Msg.String()))
+		}
+
+		conditionSet = cs
+		return nil
+	})
 	if err != nil {
-		s.logger.Audit.PolicyCRUDFailure(ctx, auditParams)
-		return nil, db.StatusifyError(ctx, s.logger, err, db.ErrTextCreationFailed, slog.String("subjectConditionSet", req.Msg.String()))
+		return nil, err
 	}
 
 	auditParams.ObjectID = conditionSet.GetId()
 	auditParams.Original = conditionSet
 	s.logger.Audit.PolicyCRUDSuccess(ctx, auditParams)
 
 	rsp.SubjectConditionSet = conditionSet
+
 	return connect.NewResponse(rsp), nil
 }
 
@@ -294,25 +304,36 @@ func (s SubjectMappingService) UpdateSubjectConditionSet(ctx context.Context,
 		ObjectID:   subjectConditionSetID,
 	}
 
-	original, err := s.dbClient.GetSubjectConditionSet(ctx, subjectConditionSetID)
-	if err != nil {
-		s.logger.Audit.PolicyCRUDFailure(ctx, auditParams)
-		return nil, db.StatusifyError(ctx, s.logger, err, db.ErrTextGetRetrievalFailed, slog.String("id", subjectConditionSetID))
-	}
+	var original, updated *policy.SubjectConditionSet
+	err := s.dbClient.RunInTx(ctx, func(txClient *policydb.PolicyDBClient) error {
+		orig, err := txClient.GetSubjectConditionSet(ctx, subjectConditionSetID)
+		if err != nil {
+			s.logger.Audit.PolicyCRUDFailure(ctx, auditParams)
+			return db.StatusifyError(ctx, s.logger, err, db.ErrTextGetRetrievalFailed, slog.String("id", subjectConditionSetID))
+		}
+
+		upd, err := txClient.UpdateSubjectConditionSet(ctx, req.Msg)
+		if err != nil {
+			s.logger.Audit.PolicyCRUDFailure(ctx, auditParams)
+			return db.StatusifyError(ctx, s.logger, err, db.ErrTextUpdateFailed, slog.String("id", req.Msg.GetId()), slog.String("subjectConditionSet fields", req.Msg.String()))
+		}
+
+		original = orig
+		updated = upd
+		rsp.SubjectConditionSet = &policy.SubjectConditionSet{
+			Id: subjectConditionSetID,
+		}
 
-	updated, err := s.dbClient.UpdateSubjectConditionSet(ctx, req.Msg)
+		return nil
+	})
 	if err != nil {
-		s.logger.Audit.PolicyCRUDFailure(ctx, auditParams)
-		return nil, db.StatusifyError(ctx, s.logger, err, db.ErrTextUpdateFailed, slog.String("id", req.Msg.GetId()), slog.String("subjectConditionSet fields", req.Msg.String()))
+		return nil, err
 	}
 
 	auditParams.Original = original
 	auditParams.Updated = updated
 	s.logger.Audit.PolicyCRUDSuccess(ctx, auditParams)
 
-	rsp.SubjectConditionSet = &policy.SubjectConditionSet{
-		Id: subjectConditionSetID,
-	}
 	return connect.NewResponse(rsp), nil
 }