From 27de27ac6496e647c59866d9ba57de61ac356771 Mon Sep 17 00:00:00 2001 From: wucm667 Date: Tue, 29 Sep 2026 23:24:45 +0800 Subject: [PATCH] fix(risk-control): remove deleted groups from moderation scope --- backend/internal/repository/group_repo.go | 23 ++++ .../repository/group_repo_integration_test.go | 127 ++++++++++++++++++ backend/internal/repository/setting_repo.go | 29 ++++ .../internal/service/content_moderation.go | 13 +- 4 files changed, 191 insertions(+), 1 deletion(-) diff --git a/backend/internal/repository/group_repo.go b/backend/internal/repository/group_repo.go index 8b2bbb827d..1e2f7e0a59 100644 --- a/backend/internal/repository/group_repo.go +++ b/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 { diff --git a/backend/internal/repository/group_repo_integration_test.go b/backend/internal/repository/group_repo_integration_test.go index 177fbe7f17..87fb51780b 100644 --- a/backend/internal/repository/group_repo_integration_test.go +++ b/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)) diff --git a/backend/internal/repository/setting_repo.go b/backend/internal/repository/setting_repo.go index a4550e6020..815daa54be 100644 --- a/backend/internal/repository/setting_repo.go +++ b/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 } diff --git a/backend/internal/service/content_moderation.go b/backend/internal/service/content_moderation.go index 3c6df48ed2..5cb48ea0e3 100644 --- a/backend/internal/service/content_moderation.go +++ b/backend/internal/service/content_moderation.go @@ -719,7 +719,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)