Merge pull request #7711 from wucm667/fix/issue-7709-risk-control-groups

fix(risk-control): remove deleted groups from moderation scope

backend/internal/repository/group_repo.go

@@ -957,6 +957,29 @@ func (r *groupRepository) deleteCascade(ctx context.Context, id int64, requireEm
 	if _, err := txClient.Group.Delete().Where(group.IDEQ(id)).Exec(ctx); err != nil {
 		return nil, err
 	}
+	// Lock and validate the text value before using JSONB operators. A legacy
+	// malformed setting must not prevent deletion or be silently replaced.
+	var moderationValue string
+	err = scanSingleRow(ctx, exec, `SELECT value FROM settings WHERE key = $1 FOR UPDATE`,
+		[]any{service.SettingKeyContentModerationConfig}, &moderationValue)
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return nil, err
+	}
+	// Keep the moderation scope in sync in the same transaction. Updating the
+	// setting row in SQL also serializes concurrent group deletions across nodes.
+	if err == nil && json.Valid([]byte(moderationValue)) {
+		if _, err := exec.ExecContext(ctx, `UPDATE settings
+		SET value = jsonb_set(value::jsonb, '{group_ids}',
+			(SELECT COALESCE(jsonb_agg(group_id ORDER BY ord), '[]'::jsonb)
+			 FROM jsonb_array_elements(value::jsonb->'group_ids') WITH ORDINALITY AS ids(group_id, ord)
+			 WHERE group_id <> to_jsonb($2::bigint)))::text,
+			updated_at = NOW()
+		WHERE key = $1 AND jsonb_typeof(value::jsonb->'group_ids') = 'array'
+			AND value::jsonb->'group_ids' @> jsonb_build_array($2::bigint)`,
+			service.SettingKeyContentModerationConfig, id); err != nil {
+			return nil, err
+		}
+	}
 
 	if tx != nil {
 		if err := tx.Commit(); err != nil {

backend/internal/repository/group_repo_integration_test.go

@@ -5,7 +5,9 @@ package repository
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 	"errors"
+	"fmt"
 	"strings"
 	"testing"
 	"time"
@@ -49,6 +51,55 @@ func TestGroupRepoSuite(t *testing.T) {
 	suite.Run(t, new(GroupRepoSuite))
 }
 
+func TestModerationConfigWriteRejectsGroupDeletedAfterValidation(t *testing.T) {
+	ctx := context.Background()
+	client := testEntClient(t)
+	repo := newGroupRepositoryWithSQL(client, integrationDB)
+	settings := &settingRepository{client: client}
+	group := &service.Group{Name: "moderation-concurrent-delete", Platform: service.PlatformAnthropic, RateMultiplier: 1, Status: service.StatusActive, SubscriptionType: service.SubscriptionTypeStandard}
+	require.NoError(t, repo.Create(ctx, group))
+	t.Cleanup(func() { _, _ = integrationDB.ExecContext(ctx, "DELETE FROM groups WHERE id=$1", group.ID) })
+	key := service.SettingKeyContentModerationConfig
+	previous, previousErr := settings.GetValue(ctx, key)
+	t.Cleanup(func() {
+		if previousErr == nil {
+			_ = settings.Set(ctx, key, previous)
+		} else {
+			_ = settings.Delete(ctx, key)
+		}
+	})
+	require.NoError(t, settings.Set(ctx, key, `{"all_groups":false,"group_ids":[]}`))
+
+	// Hold the deleter's group lock while the config writer starts with an ID
+	// that was valid before deletion. It must recheck after this transaction commits.
+	tx, err := integrationDB.BeginTx(ctx, nil)
+	require.NoError(t, err)
+	defer func() { _ = tx.Rollback() }()
+	var id int64
+	require.NoError(t, tx.QueryRowContext(ctx, "SELECT id FROM groups WHERE id=$1 FOR UPDATE", group.ID).Scan(&id))
+	finished := make(chan error, 1)
+	go func() {
+		finished <- settings.SetContentModerationConfig(ctx, `{"all_groups":false,"group_ids":[`+fmt.Sprint(group.ID)+`]}`, []int64{group.ID})
+	}()
+	select {
+	case err := <-finished:
+		t.Fatalf("config write passed a locked group before deletion: %v", err)
+	case <-time.After(100 * time.Millisecond):
+	}
+	_, err = tx.ExecContext(ctx, "UPDATE groups SET deleted_at=NOW() WHERE id=$1", group.ID)
+	require.NoError(t, err)
+	require.NoError(t, tx.Commit())
+	select {
+	case err := <-finished:
+		require.ErrorIs(t, err, service.ErrGroupNotFound)
+	case <-time.After(5 * time.Second):
+		t.Fatal("config write did not finish after deletion")
+	}
+	stored, err := settings.GetValue(ctx, key)
+	require.NoError(t, err)
+	require.JSONEq(t, `{"all_groups":false,"group_ids":[]}`, stored)
+}
+
 // --- Create / GetByID / Update / Delete ---
 
 func (s *GroupRepoSuite) TestCreate() {
@@ -257,8 +308,14 @@ func (s *GroupRepoSuite) TestDeleteCascadeIfEmptyRejectsGroupWithNonDeletedAccou
 	_, err := s.tx.ExecContext(s.ctx, "INSERT INTO account_groups (account_id, group_id, priority, created_at) VALUES ($1, $2, 1, NOW())", accountID, group.ID)
 	s.Require().NoError(err)
 
+	config, marshalErr := json.Marshal(map[string]any{"group_ids": []int64{group.ID}, "all_groups": false})
+	s.Require().NoError(marshalErr)
+	s.Require().NoError(NewSettingRepository(s.tx.Client()).Set(s.ctx, service.SettingKeyContentModerationConfig, string(config)))
 	_, err = s.repo.DeleteCascadeIfEmpty(s.ctx, group.ID)
 	s.Require().ErrorIs(err, service.ErrGroupNotEmpty)
+	stored, readErr := NewSettingRepository(s.tx.Client()).GetValue(s.ctx, service.SettingKeyContentModerationConfig)
+	s.Require().NoError(readErr)
+	s.Require().JSONEq(string(config), stored)
 	_, err = s.repo.GetByID(s.ctx, group.ID)
 	s.Require().NoError(err)
 	var bindings int
@@ -276,6 +333,76 @@ func (s *GroupRepoSuite) TestDeleteCascadeIfEmptyDeletesEmptyGroup() {
 	s.Require().ErrorIs(err, service.ErrGroupNotFound)
 }
 
+func (s *GroupRepoSuite) TestDeleteCascadeRemovesOnlySelectedModerationGroup() {
+	for _, guarded := range []bool{false, true} {
+		name := map[bool]string{false: "normal", true: "guarded"}[guarded]
+		s.Run(name, func() {
+			deleted := &service.Group{Name: "moderation-deleted-" + name, Platform: service.PlatformAnthropic, RateMultiplier: 1, Status: service.StatusActive, SubscriptionType: service.SubscriptionTypeStandard}
+			kept := &service.Group{Name: "moderation-kept-" + name, Platform: service.PlatformAnthropic, RateMultiplier: 1, Status: service.StatusActive, SubscriptionType: service.SubscriptionTypeStandard}
+			s.Require().NoError(s.repo.Create(s.ctx, deleted))
+			s.Require().NoError(s.repo.Create(s.ctx, kept))
+			config := map[string]any{"group_ids": []int64{kept.ID, deleted.ID}, "all_groups": false, "mode": "observe"}
+			raw, err := json.Marshal(config)
+			s.Require().NoError(err)
+			s.Require().NoError(NewSettingRepository(s.tx.Client()).Set(s.ctx, service.SettingKeyContentModerationConfig, string(raw)))
+			if guarded {
+				_, err = s.repo.DeleteCascadeIfEmpty(s.ctx, deleted.ID)
+			} else {
+				_, err = s.repo.DeleteCascade(s.ctx, deleted.ID)
+			}
+			s.Require().NoError(err)
+			stored, err := NewSettingRepository(s.tx.Client()).GetValue(s.ctx, service.SettingKeyContentModerationConfig)
+			s.Require().NoError(err)
+			var got struct {
+				GroupIDs []int64 `json:"group_ids"`
+				Mode     string  `json:"mode"`
+			}
+			s.Require().NoError(json.Unmarshal([]byte(stored), &got))
+			s.Require().Equal([]int64{kept.ID}, got.GroupIDs)
+			s.Require().Equal("observe", got.Mode)
+		})
+	}
+}
+
+func (s *GroupRepoSuite) TestDeleteCascadeUnknownGroupPreservesModerationConfig() {
+	const original = `{"all_groups":false,"group_ids":[123,456],"mode":"observe"}`
+	s.Require().NoError(NewSettingRepository(s.tx.Client()).Set(s.ctx, service.SettingKeyContentModerationConfig, original))
+	for _, guarded := range []bool{false, true} {
+		var err error
+		if guarded {
+			_, err = s.repo.DeleteCascadeIfEmpty(s.ctx, -1)
+		} else {
+			_, err = s.repo.DeleteCascade(s.ctx, -1)
+		}
+		s.Require().ErrorIs(err, service.ErrGroupNotFound)
+		stored, readErr := NewSettingRepository(s.tx.Client()).GetValue(s.ctx, service.SettingKeyContentModerationConfig)
+		s.Require().NoError(readErr)
+		s.Require().JSONEq(original, stored)
+	}
+}
+
+func (s *GroupRepoSuite) TestDeleteCascadePreservesMalformedModerationConfig() {
+	for _, guarded := range []bool{false, true} {
+		name := map[bool]string{false: "normal", true: "guarded"}[guarded]
+		s.Run(name, func() {
+			group := &service.Group{Name: "malformed-moderation-" + name, Platform: service.PlatformAnthropic, RateMultiplier: 1, Status: service.StatusActive, SubscriptionType: service.SubscriptionTypeStandard}
+			s.Require().NoError(s.repo.Create(s.ctx, group))
+			const malformed = `{"group_ids":[oops]}`
+			s.Require().NoError(NewSettingRepository(s.tx.Client()).Set(s.ctx, service.SettingKeyContentModerationConfig, malformed))
+			var err error
+			if guarded {
+				_, err = s.repo.DeleteCascadeIfEmpty(s.ctx, group.ID)
+			} else {
+				_, err = s.repo.DeleteCascade(s.ctx, group.ID)
+			}
+			s.Require().NoError(err)
+			stored, err := NewSettingRepository(s.tx.Client()).GetValue(s.ctx, service.SettingKeyContentModerationConfig)
+			s.Require().NoError(err)
+			s.Require().Equal(malformed, stored)
+		})
+	}
+}
+
 func (s *GroupRepoSuite) TestDeleteCascadeIfEmptyIgnoresBindingsToSoftDeletedAccounts() {
 	group := &service.Group{Name: "guarded-soft-deleted-account", Platform: service.PlatformAnthropic, RateMultiplier: 1, Status: service.StatusActive, SubscriptionType: service.SubscriptionTypeStandard}
 	s.Require().NoError(s.repo.Create(s.ctx, group))

backend/internal/repository/setting_repo.go

@@ -2,6 +2,7 @@ package repository
 
 import (
 	"context"
+	"errors"
 	"time"
 
 	"github.com/Wei-Shaw/sub2api/ent"
@@ -9,6 +10,34 @@ import (
 	"github.com/Wei-Shaw/sub2api/internal/service"
 )
 
+// SetContentModerationConfig holds live group rows until the setting is written.
+// Deletion takes FOR UPDATE on the group before cleaning this setting, so both
+// operations follow the same lock order: groups, then settings.
+func (r *settingRepository) SetContentModerationConfig(ctx context.Context, value string, groupIDs []int64) error {
+	tx, err := r.client.Tx(ctx)
+	if err != nil && !errors.Is(err, ent.ErrTxStarted) {
+		return err
+	}
+	client := r.client
+	if err == nil {
+		defer func() { _ = tx.Rollback() }()
+		client = tx.Client()
+	}
+	if err := lockLiveGroups(ctx, client, groupIDs); err != nil {
+		return err
+	}
+	now := time.Now()
+	if err := client.Setting.Create().SetKey(service.SettingKeyContentModerationConfig).
+		SetValue(value).SetUpdatedAt(now).OnConflictColumns(setting.FieldKey).
+		UpdateNewValues().Exec(ctx); err != nil {
+		return err
+	}
+	if tx != nil {
+		return tx.Commit()
+	}
+	return nil
+}
+
 type settingRepository struct {
 	client *ent.Client
 }

backend/internal/service/content_moderation.go

@@ -720,7 +720,18 @@ func (s *ContentModerationService) UpdateConfig(ctx context.Context, input Updat
 	if err != nil {
 		return nil, fmt.Errorf("marshal content moderation config: %w", err)
 	}
-	if err := s.settingRepo.Set(ctx, SettingKeyContentModerationConfig, string(raw)); err != nil {
+	var saveErr error
+	if writer, ok := s.settingRepo.(interface {
+		SetContentModerationConfig(context.Context, string, []int64) error
+	}); ok {
+		saveErr = writer.SetContentModerationConfig(ctx, string(raw), cfg.GroupIDs)
+		if errors.Is(saveErr, ErrGroupNotFound) {
+			return nil, infraerrors.BadRequest("INVALID_CONTENT_MODERATION_GROUP", "审计分组不存在")
+		}
+	} else {
+		saveErr = s.settingRepo.Set(ctx, SettingKeyContentModerationConfig, string(raw))
+	}
+	if err := saveErr; err != nil {
 		return nil, fmt.Errorf("save content moderation config: %w", err)
 	}
 	s.replaceRuntimeConfig(cfg, raw)