From 8b37ba8828d77ada5d05e90cac665d9ac5532d60 Mon Sep 17 00:00:00 2001 From: Lyonle <214648221+lyon-le@users.noreply.github.com> Date: Tue, 14 Jul 2026 16:34:20 +0800 Subject: [PATCH] perf(risk): optimize keyword moderation hot path --- .../internal/service/content_moderation.go | 184 ++++++- .../content_moderation_keyword_matcher.go | 222 +++++++++ ...content_moderation_keyword_matcher_test.go | 59 +++ .../content_moderation_runtime_cache_test.go | 451 ++++++++++++++++++ 4 files changed, 900 insertions(+), 16 deletions(-) create mode 100644 backend/internal/service/content_moderation_keyword_matcher.go create mode 100644 backend/internal/service/content_moderation_keyword_matcher_test.go create mode 100644 backend/internal/service/content_moderation_runtime_cache_test.go diff --git a/backend/internal/service/content_moderation.go b/backend/internal/service/content_moderation.go index f633c8ad17..8102f34a90 100644 --- a/backend/internal/service/content_moderation.go +++ b/backend/internal/service/content_moderation.go @@ -92,6 +92,9 @@ const ( contentModerationCleanupInterval = 24 * time.Hour contentModerationCleanupTimeout = 30 * time.Minute contentModerationCleanupDelay = 5 * time.Minute + + contentModerationRuntimeCacheTTL = time.Second + contentModerationRuntimeRefreshTimeout = 5 * time.Second ) var contentModerationCategoryOrder = []string{ @@ -512,10 +515,22 @@ type ContentModerationService struct { lastCleanupUnix atomic.Int64 lastCleanupDeletedHit atomic.Int64 lastCleanupDeletedNonHit atomic.Int64 + runtimeSnapshot atomic.Pointer[contentModerationRuntimeSnapshot] + runtimeRefreshMu sync.Mutex + runtimeCacheTTL time.Duration + runtimeRefreshRetryAt atomic.Int64 keyHealthMu sync.Mutex keyHealth map[string]*contentModerationKeyHealth } +type contentModerationRuntimeSnapshot struct { + riskControlEnabled bool + config *ContentModerationConfig + keywordMatcher *contentModerationKeywordMatcher + configDigest [sha256.Size]byte + loadedAt time.Time +} + type contentModerationTask struct { input ContentModerationCheckInput content ContentModerationInput @@ -700,6 +715,7 @@ func (s *ContentModerationService) UpdateConfig(ctx context.Context, input Updat if err := s.settingRepo.Set(ctx, SettingKeyContentModerationConfig, string(raw)); err != nil { return nil, fmt.Errorf("save content moderation config: %w", err) } + s.replaceRuntimeConfig(cfg, raw) return s.configView(cfg), nil } @@ -776,16 +792,7 @@ func (s *ContentModerationService) Check(ctx context.Context, input ContentModer "protocol", input.Protocol) return allow, nil } - if !s.isRiskControlEnabled(ctx) { - slog.Info("content_moderation.skip_feature_disabled", - "user_id", input.UserID, - "api_key_id", input.APIKeyID, - "group_id", contentModerationLogGroupID(input.GroupID), - "endpoint", input.Endpoint, - "protocol", input.Protocol) - return allow, nil - } - cfg, err := s.loadConfig(ctx) + runtimeSnapshot, err := s.loadRuntimeSnapshot(ctx) if err != nil { slog.Warn("content_moderation.skip_config_load_failed", "user_id", input.UserID, @@ -796,6 +803,16 @@ func (s *ContentModerationService) Check(ctx context.Context, input ContentModer "error", err) return allow, nil } + if !runtimeSnapshot.riskControlEnabled { + slog.Info("content_moderation.skip_feature_disabled", + "user_id", input.UserID, + "api_key_id", input.APIKeyID, + "group_id", contentModerationLogGroupID(input.GroupID), + "endpoint", input.Endpoint, + "protocol", input.Protocol) + return allow, nil + } + cfg := runtimeSnapshot.config inGroupScope := cfg.includesGroup(input.GroupID) inModelScope := cfg.includesModel(input.Model) slog.Info("content_moderation.config_loaded", @@ -885,7 +902,7 @@ func (s *ContentModerationService) Check(ctx context.Context, input ContentModer hashText := content.Hash() if cfg.Mode == ContentModerationModePreBlock { if cfg.KeywordBlockingMode != ContentModerationKeywordModeAPIOnly && len(cfg.BlockedKeywords) > 0 { - if keyword, hit := matchBlockedKeyword(content.Text, cfg.BlockedKeywords); hit { + if keyword, hit := runtimeSnapshot.matchBlockedKeyword(content.Text); hit { s.recordPreBlockSyncMetric(0, ContentModerationActionKeywordBlock) slog.Info("content_moderation.keyword_block", "user_id", input.UserID, @@ -1178,12 +1195,13 @@ func (s *ContentModerationService) enqueueRecord(input ContentModerationCheckInp func (s *ContentModerationService) worker(id int) { for { ctx, cancel := context.WithTimeout(context.Background(), maxContentModerationTimeoutMS*time.Millisecond+10*time.Second) - cfg, err := s.loadConfig(ctx) - if err != nil || id >= cfg.WorkerCount { + runtimeSnapshot, err := s.loadRuntimeSnapshot(ctx) + if err != nil || runtimeSnapshot == nil || runtimeSnapshot.config == nil || id >= runtimeSnapshot.config.WorkerCount { cancel() time.Sleep(time.Second) continue } + cfg := runtimeSnapshot.config task, ok := s.dequeueAsyncTask(ctx, time.Second) if !ok { cancel() @@ -1438,15 +1456,18 @@ func (s *ContentModerationService) runCleanupOnce() { } func (s *ContentModerationService) loadConfig(ctx context.Context) (*ContentModerationConfig, error) { - cfg := defaultContentModerationConfig() raw, err := s.settingRepo.GetValue(ctx, SettingKeyContentModerationConfig) if err != nil { if errors.Is(err, ErrSettingNotFound) { - cfg.normalize() - return cfg, nil + return parseContentModerationConfig("") } return nil, fmt.Errorf("get content moderation config: %w", err) } + return parseContentModerationConfig(raw) +} + +func parseContentModerationConfig(raw string) (*ContentModerationConfig, error) { + cfg := defaultContentModerationConfig() if strings.TrimSpace(raw) == "" { cfg.normalize() return cfg, nil @@ -1458,6 +1479,137 @@ func (s *ContentModerationService) loadConfig(ctx context.Context) (*ContentMode return cfg, nil } +func (s *ContentModerationService) loadRuntimeSnapshot(ctx context.Context) (*contentModerationRuntimeSnapshot, error) { + if s == nil || s.settingRepo == nil { + return nil, errors.New("content moderation setting repository unavailable") + } + now := time.Now() + if snapshot := s.runtimeSnapshot.Load(); snapshot != nil { + if now.Sub(snapshot.loadedAt) < s.runtimeSnapshotTTL() { + return snapshot, nil + } + s.triggerRuntimeSnapshotRefresh() + return snapshot, nil + } + + s.runtimeRefreshMu.Lock() + defer s.runtimeRefreshMu.Unlock() + if snapshot := s.runtimeSnapshot.Load(); snapshot != nil { + return snapshot, nil + } + return s.refreshRuntimeSnapshot(ctx) +} + +func (s *ContentModerationService) runtimeSnapshotTTL() time.Duration { + if s != nil && s.runtimeCacheTTL > 0 { + return s.runtimeCacheTTL + } + return contentModerationRuntimeCacheTTL +} + +func (s *ContentModerationService) triggerRuntimeSnapshotRefresh() { + if s == nil || s.runtimeRefreshDeferred() || !s.runtimeRefreshMu.TryLock() { + return + } + if s.runtimeRefreshDeferred() { + s.runtimeRefreshMu.Unlock() + return + } + go func() { + defer s.runtimeRefreshMu.Unlock() + ctx, cancel := context.WithTimeout(context.Background(), contentModerationRuntimeRefreshTimeout) + defer cancel() + if _, err := s.refreshRuntimeSnapshot(ctx); err != nil { + s.runtimeRefreshRetryAt.Store(time.Now().Add(s.runtimeSnapshotTTL()).UnixNano()) + slog.Warn("content_moderation.runtime_snapshot_refresh_failed", "error", err) + } + }() +} + +func (s *ContentModerationService) runtimeRefreshDeferred() bool { + if s == nil { + return false + } + return time.Now().UnixNano() < s.runtimeRefreshRetryAt.Load() +} + +func (s *ContentModerationService) refreshRuntimeSnapshot(ctx context.Context) (*contentModerationRuntimeSnapshot, error) { + values, err := s.settingRepo.GetMultiple(ctx, []string{ + SettingKeyRiskControlEnabled, + SettingKeyContentModerationConfig, + }) + if err != nil { + return nil, fmt.Errorf("get content moderation runtime settings: %w", err) + } + rawConfig := values[SettingKeyContentModerationConfig] + configDigest := sha256.Sum256([]byte(rawConfig)) + if current := s.runtimeSnapshot.Load(); current != nil && current.configDigest == configDigest { + snapshot := &contentModerationRuntimeSnapshot{ + riskControlEnabled: values[SettingKeyRiskControlEnabled] == "true", + config: current.config, + keywordMatcher: current.keywordMatcher, + configDigest: configDigest, + loadedAt: time.Now(), + } + s.runtimeSnapshot.Store(snapshot) + s.runtimeRefreshRetryAt.Store(0) + return snapshot, nil + } + cfg, err := parseContentModerationConfig(rawConfig) + if err != nil { + return nil, err + } + snapshot := &contentModerationRuntimeSnapshot{ + riskControlEnabled: values[SettingKeyRiskControlEnabled] == "true", + config: cfg, + keywordMatcher: newContentModerationKeywordMatcher(cfg.BlockedKeywords), + configDigest: configDigest, + loadedAt: time.Now(), + } + s.runtimeSnapshot.Store(snapshot) + s.runtimeRefreshRetryAt.Store(0) + return snapshot, nil +} + +func (s *ContentModerationService) replaceRuntimeConfig(cfg *ContentModerationConfig, raw []byte) { + if s == nil || cfg == nil { + return + } + s.runtimeRefreshMu.Lock() + hasSnapshot := s.runtimeSnapshot.Load() != nil + s.runtimeRefreshMu.Unlock() + if !hasSnapshot { + return + } + config := cloneContentModerationConfig(cfg) + keywordMatcher := newContentModerationKeywordMatcher(cfg.BlockedKeywords) + configDigest := sha256.Sum256(raw) + + s.runtimeRefreshMu.Lock() + defer s.runtimeRefreshMu.Unlock() + current := s.runtimeSnapshot.Load() + if current == nil { + return + } + s.runtimeSnapshot.Store(&contentModerationRuntimeSnapshot{ + riskControlEnabled: current.riskControlEnabled, + config: config, + keywordMatcher: keywordMatcher, + configDigest: configDigest, + loadedAt: time.Now(), + }) +} + +func (s *contentModerationRuntimeSnapshot) matchBlockedKeyword(text string) (string, bool) { + if s == nil || s.config == nil { + return "", false + } + if s.keywordMatcher != nil { + return s.keywordMatcher.Match(text) + } + return matchBlockedKeyword(text, s.config.BlockedKeywords) +} + func (s *ContentModerationService) isRiskControlEnabled(ctx context.Context) bool { raw, err := s.settingRepo.GetValue(ctx, SettingKeyRiskControlEnabled) if err != nil { diff --git a/backend/internal/service/content_moderation_keyword_matcher.go b/backend/internal/service/content_moderation_keyword_matcher.go new file mode 100644 index 0000000000..4628d3b70a --- /dev/null +++ b/backend/internal/service/content_moderation_keyword_matcher.go @@ -0,0 +1,222 @@ +package service + +import ( + "strings" +) + +type contentModerationKeywordMatcher struct { + nodes []contentModerationKeywordNode + edges []contentModerationKeywordEdge + rootTransitions [256]int32 + keywords []string +} + +type contentModerationKeywordNode struct { + failure int32 + bestKeyword int32 + edgeStart uint32 + edgeCount uint16 +} + +type contentModerationKeywordEdge struct { + target int32 + label byte +} + +type contentModerationKeywordBuildEdge struct { + target int32 + nextSibling int32 + label byte +} + +func newContentModerationKeywordMatcher(keywords []string) *contentModerationKeywordMatcher { + if len(keywords) == 0 { + return nil + } + + buildNodes := []contentModerationKeywordNode{newContentModerationKeywordNode()} + buildEdges := make([]contentModerationKeywordBuildEdge, 0) + originalKeywords := append([]string(nil), keywords...) + + for keywordIndex, keyword := range keywords { + if keyword == "" { + continue + } + state := int32(0) + for _, label := range []byte(strings.ToLower(keyword)) { + next := contentModerationKeywordBuildTransition(buildNodes, buildEdges, state, label) + if next < 0 { + next = int32(len(buildNodes)) + buildNodes = append(buildNodes, newContentModerationKeywordNode()) + buildEdges = append(buildEdges, contentModerationKeywordBuildEdge{ + target: next, + nextSibling: contentModerationKeywordBuildFirstEdge(buildNodes[state]), + label: label, + }) + buildNodes[state].edgeStart = uint32(len(buildEdges)) + } + state = next + } + if current := buildNodes[state].bestKeyword; current < 0 || int32(keywordIndex) < current { + buildNodes[state].bestKeyword = int32(keywordIndex) + } + } + + if len(buildNodes) == 1 { + return nil + } + + queue := make([]int32, 0, len(buildNodes)-1) + var rootTransitions [256]int32 + for edgeIndex := contentModerationKeywordBuildFirstEdge(buildNodes[0]); edgeIndex >= 0; edgeIndex = buildEdges[edgeIndex].nextSibling { + edge := buildEdges[edgeIndex] + rootTransitions[edge.label] = edge.target + queue = append(queue, edge.target) + } + + for queueIndex := 0; queueIndex < len(queue); queueIndex++ { + state := queue[queueIndex] + for edgeIndex := contentModerationKeywordBuildFirstEdge(buildNodes[state]); edgeIndex >= 0; edgeIndex = buildEdges[edgeIndex].nextSibling { + edge := buildEdges[edgeIndex] + failure := buildNodes[state].failure + fallback := contentModerationKeywordBuildTransition(buildNodes, buildEdges, failure, edge.label) + for fallback < 0 && failure != 0 { + failure = buildNodes[failure].failure + fallback = contentModerationKeywordBuildTransition(buildNodes, buildEdges, failure, edge.label) + } + if fallback >= 0 { + buildNodes[edge.target].failure = fallback + } + buildNodes[edge.target].bestKeyword = minKeywordIndex( + buildNodes[edge.target].bestKeyword, + buildNodes[buildNodes[edge.target].failure].bestKeyword, + ) + queue = append(queue, edge.target) + } + } + + edges := make([]contentModerationKeywordEdge, 0, len(buildEdges)) + var outgoing [256]contentModerationKeywordEdge + for nodeIndex := range buildNodes { + count := 0 + for edgeIndex := contentModerationKeywordBuildFirstEdge(buildNodes[nodeIndex]); edgeIndex >= 0; edgeIndex = buildEdges[edgeIndex].nextSibling { + edge := buildEdges[edgeIndex] + outgoing[count] = contentModerationKeywordEdge{target: edge.target, label: edge.label} + count++ + } + for index := 1; index < count; index++ { + current := outgoing[index] + insertAt := index + for insertAt > 0 && current.label < outgoing[insertAt-1].label { + outgoing[insertAt] = outgoing[insertAt-1] + insertAt-- + } + outgoing[insertAt] = current + } + buildNodes[nodeIndex].edgeStart = uint32(len(edges)) + buildNodes[nodeIndex].edgeCount = uint16(count) + edges = append(edges, outgoing[:count]...) + } + + return &contentModerationKeywordMatcher{ + nodes: buildNodes, + edges: edges, + rootTransitions: rootTransitions, + keywords: originalKeywords, + } +} + +func newContentModerationKeywordNode() contentModerationKeywordNode { + return contentModerationKeywordNode{bestKeyword: -1} +} + +func contentModerationKeywordBuildFirstEdge(node contentModerationKeywordNode) int32 { + if node.edgeStart == 0 { + return -1 + } + return int32(node.edgeStart - 1) +} + +func contentModerationKeywordBuildTransition( + nodes []contentModerationKeywordNode, + edges []contentModerationKeywordBuildEdge, + state int32, + label byte, +) int32 { + if state < 0 || int(state) >= len(nodes) { + return -1 + } + for edgeIndex := contentModerationKeywordBuildFirstEdge(nodes[state]); edgeIndex >= 0; edgeIndex = edges[edgeIndex].nextSibling { + if edges[edgeIndex].label == label { + return edges[edgeIndex].target + } + } + return -1 +} + +func minKeywordIndex(left, right int32) int32 { + if left < 0 { + return right + } + if right < 0 || left < right { + return left + } + return right +} + +func (m *contentModerationKeywordMatcher) Match(text string) (string, bool) { + if m == nil || text == "" || len(m.nodes) == 0 || len(m.keywords) == 0 { + return "", false + } + lower := strings.ToLower(text) + state := int32(0) + bestKeyword := int32(-1) + for index := 0; index < len(lower); index++ { + label := lower[index] + for { + next := m.next(state, label) + if next != 0 { + state = next + break + } + if state == 0 { + break + } + state = m.nodes[state].failure + } + bestKeyword = minKeywordIndex(bestKeyword, m.nodes[state].bestKeyword) + if bestKeyword == 0 { + return m.keywords[0], true + } + } + if bestKeyword < 0 || int(bestKeyword) >= len(m.keywords) { + return "", false + } + return m.keywords[bestKeyword], true +} + +func (m *contentModerationKeywordMatcher) next(state int32, label byte) int32 { + if state == 0 { + return m.rootTransitions[label] + } + if state < 0 || int(state) >= len(m.nodes) { + return 0 + } + node := m.nodes[state] + left := int(node.edgeStart) + right := left + int(node.edgeCount) + for left < right { + middle := left + (right-left)/2 + edge := m.edges[middle] + if edge.label < label { + left = middle + 1 + continue + } + right = middle + } + end := int(node.edgeStart) + int(node.edgeCount) + if left < end && m.edges[left].label == label { + return m.edges[left].target + } + return 0 +} diff --git a/backend/internal/service/content_moderation_keyword_matcher_test.go b/backend/internal/service/content_moderation_keyword_matcher_test.go new file mode 100644 index 0000000000..4be70dca3e --- /dev/null +++ b/backend/internal/service/content_moderation_keyword_matcher_test.go @@ -0,0 +1,59 @@ +package service + +import ( + "math/rand" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestContentModerationKeywordMatcherMatchesLegacyBehavior(t *testing.T) { + tests := []struct { + name string + text string + keywords []string + }{ + {name: "miss", text: "clean prompt", keywords: []string{"blocked", "secret"}}, + {name: "case insensitive", text: "contains SECRET value", keywords: []string{"secret"}}, + {name: "configured order wins", text: "early appears before later", keywords: []string{"later", "early"}}, + {name: "overlap uses configured order", text: "abc", keywords: []string{"bc", "abc"}}, + {name: "unicode", text: "这里包含敏感词和世界", keywords: []string{"世界", "敏感词"}}, + {name: "duplicates", text: "duplicate", keywords: []string{"duplicate", "DUPLICATE"}}, + {name: "empty entries", text: "blocked", keywords: []string{"", "blocked"}}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + wantKeyword, wantHit := matchBlockedKeyword(tt.text, tt.keywords) + gotKeyword, gotHit := newContentModerationKeywordMatcher(tt.keywords).Match(tt.text) + require.Equal(t, wantHit, gotHit) + require.Equal(t, wantKeyword, gotKeyword) + }) + } +} + +func TestContentModerationKeywordMatcherRandomizedParity(t *testing.T) { + rng := rand.New(rand.NewSource(20260714)) + const alphabet = "abcXYZ" + for iteration := 0; iteration < 1000; iteration++ { + keywords := make([]string, 1+rng.Intn(30)) + for index := range keywords { + length := 1 + rng.Intn(8) + var value strings.Builder + for range length { + _ = value.WriteByte(alphabet[rng.Intn(len(alphabet))]) + } + keywords[index] = value.String() + } + var text strings.Builder + for range 20 + rng.Intn(100) { + _ = text.WriteByte(alphabet[rng.Intn(len(alphabet))]) + } + + wantKeyword, wantHit := matchBlockedKeyword(text.String(), keywords) + gotKeyword, gotHit := newContentModerationKeywordMatcher(keywords).Match(text.String()) + require.Equal(t, wantHit, gotHit, "iteration %d", iteration) + require.Equal(t, wantKeyword, gotKeyword, "iteration %d", iteration) + } +} diff --git a/backend/internal/service/content_moderation_runtime_cache_test.go b/backend/internal/service/content_moderation_runtime_cache_test.go new file mode 100644 index 0000000000..2b34d1fa57 --- /dev/null +++ b/backend/internal/service/content_moderation_runtime_cache_test.go @@ -0,0 +1,451 @@ +package service + +import ( + "context" + "encoding/json" + "errors" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +type contentModerationRuntimeSettingRepo struct { + mu sync.Mutex + values map[string]string + getValueCalls int + getMultipleCalls int + getMultipleErr error + getMultipleStart chan<- struct{} + getMultipleWait <-chan struct{} +} + +func (r *contentModerationRuntimeSettingRepo) Get(_ context.Context, key string) (*Setting, error) { + r.mu.Lock() + defer r.mu.Unlock() + value, ok := r.values[key] + if !ok { + return nil, ErrSettingNotFound + } + return &Setting{Key: key, Value: value}, nil +} + +func (r *contentModerationRuntimeSettingRepo) GetValue(_ context.Context, key string) (string, error) { + r.mu.Lock() + defer r.mu.Unlock() + r.getValueCalls++ + value, ok := r.values[key] + if !ok { + return "", ErrSettingNotFound + } + return value, nil +} + +func (r *contentModerationRuntimeSettingRepo) Set(_ context.Context, key, value string) error { + r.mu.Lock() + defer r.mu.Unlock() + if r.values == nil { + r.values = make(map[string]string) + } + r.values[key] = value + return nil +} + +func (r *contentModerationRuntimeSettingRepo) GetMultiple(_ context.Context, keys []string) (map[string]string, error) { + r.mu.Lock() + r.getMultipleCalls++ + if err := r.getMultipleErr; err != nil { + r.mu.Unlock() + return nil, err + } + out := make(map[string]string, len(keys)) + for _, key := range keys { + if value, ok := r.values[key]; ok { + out[key] = value + } + } + start := r.getMultipleStart + wait := r.getMultipleWait + r.getMultipleStart = nil + r.getMultipleWait = nil + r.mu.Unlock() + if start != nil { + start <- struct{}{} + } + if wait != nil { + <-wait + } + return out, nil +} + +func (r *contentModerationRuntimeSettingRepo) SetMultiple(_ context.Context, values map[string]string) error { + r.mu.Lock() + defer r.mu.Unlock() + if r.values == nil { + r.values = make(map[string]string) + } + for key, value := range values { + r.values[key] = value + } + return nil +} + +func (r *contentModerationRuntimeSettingRepo) GetAll(_ context.Context) (map[string]string, error) { + r.mu.Lock() + defer r.mu.Unlock() + out := make(map[string]string, len(r.values)) + for key, value := range r.values { + out[key] = value + } + return out, nil +} + +func (r *contentModerationRuntimeSettingRepo) Delete(_ context.Context, key string) error { + r.mu.Lock() + defer r.mu.Unlock() + delete(r.values, key) + return nil +} + +func (r *contentModerationRuntimeSettingRepo) calls() (getValue, getMultiple int) { + r.mu.Lock() + defer r.mu.Unlock() + return r.getValueCalls, r.getMultipleCalls +} + +func (r *contentModerationRuntimeSettingRepo) failMultiple(err error) { + r.mu.Lock() + defer r.mu.Unlock() + r.getMultipleErr = err +} + +func (r *contentModerationRuntimeSettingRepo) blockNextMultiple(start chan<- struct{}, wait <-chan struct{}) { + r.mu.Lock() + defer r.mu.Unlock() + r.getMultipleStart = start + r.getMultipleWait = wait +} + +func runtimeCacheTestConfig(t *testing.T, keywords ...string) string { + t.Helper() + cfg := defaultContentModerationConfig() + cfg.Enabled = true + cfg.Mode = ContentModerationModePreBlock + cfg.KeywordBlockingMode = ContentModerationKeywordModeKeywordOnly + cfg.BlockedKeywords = keywords + raw, err := json.Marshal(cfg) + require.NoError(t, err) + return string(raw) +} + +func runtimeCacheTestService(repo *contentModerationRuntimeSettingRepo, ttl time.Duration) *ContentModerationService { + return &ContentModerationService{ + settingRepo: repo, + repo: &contentModerationTestRepo{}, + runtimeCacheTTL: ttl, + } +} + +func runtimeCacheTestInput(text string) ContentModerationCheckInput { + return ContentModerationCheckInput{ + Protocol: ContentModerationProtocolOpenAIChat, + Model: "risk-cache-test", + Body: []byte(`{"messages":[{"role":"user","content":"` + text + `"}]}`), + } +} + +func TestContentModerationRuntimeSnapshotCachesSettings(t *testing.T) { + repo := &contentModerationRuntimeSettingRepo{values: map[string]string{ + SettingKeyRiskControlEnabled: "true", + SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "blocked"), + }} + svc := runtimeCacheTestService(repo, time.Hour) + + for range 20 { + decision, err := svc.Check(context.Background(), runtimeCacheTestInput("clean prompt")) + require.NoError(t, err) + require.True(t, decision.Allowed) + } + + getValue, getMultiple := repo.calls() + require.Zero(t, getValue) + require.Equal(t, 1, getMultiple) +} + +func TestContentModerationRuntimeSnapshotUpdateConfigIsImmediate(t *testing.T) { + repo := &contentModerationRuntimeSettingRepo{values: map[string]string{ + SettingKeyRiskControlEnabled: "true", + SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "old-keyword"), + }} + svc := runtimeCacheTestService(repo, time.Hour) + + decision, err := svc.Check(context.Background(), runtimeCacheTestInput("new-keyword")) + require.NoError(t, err) + require.True(t, decision.Allowed) + + keywords := []string{"new-keyword"} + _, err = svc.UpdateConfig(context.Background(), UpdateContentModerationConfigInput{ + BlockedKeywords: &keywords, + }) + require.NoError(t, err) + + decision, err = svc.Check(context.Background(), runtimeCacheTestInput("new-keyword")) + require.NoError(t, err) + require.True(t, decision.Blocked) + + _, getMultiple := repo.calls() + require.Equal(t, 1, getMultiple) +} + +func TestContentModerationRuntimeSnapshotUpdateWinsOverInitialLoad(t *testing.T) { + repo := &contentModerationRuntimeSettingRepo{values: map[string]string{ + SettingKeyRiskControlEnabled: "true", + SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "old-keyword"), + }} + svc := runtimeCacheTestService(repo, time.Hour) + + refreshStarted := make(chan struct{}, 1) + releaseRefresh := make(chan struct{}) + released := false + defer func() { + if !released { + close(releaseRefresh) + } + }() + repo.blockNextMultiple(refreshStarted, releaseRefresh) + + initialCheckDone := make(chan error, 1) + go func() { + decision, err := svc.Check(context.Background(), runtimeCacheTestInput("clean prompt")) + if err == nil && (decision == nil || !decision.Allowed) { + err = errors.New("unexpected initial moderation decision") + } + initialCheckDone <- err + }() + require.Eventually(t, func() bool { + select { + case <-refreshStarted: + return true + default: + return false + } + }, time.Second, time.Millisecond) + + updateDone := make(chan error, 1) + go func() { + keywords := []string{"new-keyword"} + _, updateErr := svc.UpdateConfig(context.Background(), UpdateContentModerationConfigInput{ + BlockedKeywords: &keywords, + }) + updateDone <- updateErr + }() + select { + case updateErr := <-updateDone: + require.NoError(t, updateErr) + t.Fatal("configuration update completed before the initial load released its lock") + case <-time.After(10 * time.Millisecond): + } + + close(releaseRefresh) + released = true + require.NoError(t, <-initialCheckDone) + require.NoError(t, <-updateDone) + + decision, err := svc.Check(context.Background(), runtimeCacheTestInput("new-keyword")) + require.NoError(t, err) + require.True(t, decision.Blocked) + decision, err = svc.Check(context.Background(), runtimeCacheTestInput("old-keyword")) + require.NoError(t, err) + require.True(t, decision.Allowed) +} + +func TestContentModerationRuntimeSnapshotRefreshFailureKeepsStaleConfig(t *testing.T) { + repo := &contentModerationRuntimeSettingRepo{values: map[string]string{ + SettingKeyRiskControlEnabled: "true", + SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "blocked"), + }} + svc := runtimeCacheTestService(repo, time.Nanosecond) + input := runtimeCacheTestInput("blocked") + + decision, err := svc.Check(context.Background(), input) + require.NoError(t, err) + require.True(t, decision.Blocked) + + repo.failMultiple(errors.New("database unavailable")) + decision, err = svc.Check(context.Background(), input) + require.NoError(t, err) + require.True(t, decision.Blocked) + require.Eventually(t, func() bool { + _, calls := repo.calls() + return calls >= 2 + }, time.Second, time.Millisecond) +} + +func TestContentModerationRuntimeSnapshotRefreshFailureBacksOff(t *testing.T) { + repo := &contentModerationRuntimeSettingRepo{values: map[string]string{ + SettingKeyRiskControlEnabled: "true", + SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "blocked"), + }} + svc := runtimeCacheTestService(repo, time.Minute) + input := runtimeCacheTestInput("blocked") + + decision, err := svc.Check(context.Background(), input) + require.NoError(t, err) + require.True(t, decision.Blocked) + + current := svc.runtimeSnapshot.Load() + require.NotNil(t, current) + expired := *current + expired.loadedAt = time.Now().Add(-2 * time.Minute) + svc.runtimeSnapshot.Store(&expired) + repo.failMultiple(errors.New("database unavailable")) + + decision, err = svc.Check(context.Background(), input) + require.NoError(t, err) + require.True(t, decision.Blocked) + require.Eventually(t, func() bool { + _, calls := repo.calls() + return calls == 2 + }, time.Second, time.Millisecond) + + for range 100 { + decision, err = svc.Check(context.Background(), input) + require.NoError(t, err) + require.True(t, decision.Blocked) + } + _, calls := repo.calls() + require.Equal(t, 2, calls) +} + +func TestContentModerationRuntimeSnapshotRefreshReusesUnchangedMatcher(t *testing.T) { + repo := &contentModerationRuntimeSettingRepo{values: map[string]string{ + SettingKeyRiskControlEnabled: "true", + SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "blocked"), + }} + svc := runtimeCacheTestService(repo, time.Minute) + input := runtimeCacheTestInput("blocked") + + decision, err := svc.Check(context.Background(), input) + require.NoError(t, err) + require.True(t, decision.Blocked) + + current := svc.runtimeSnapshot.Load() + require.NotNil(t, current) + expired := *current + expired.loadedAt = time.Now().Add(-2 * time.Minute) + svc.runtimeSnapshot.Store(&expired) + + decision, err = svc.Check(context.Background(), input) + require.NoError(t, err) + require.True(t, decision.Blocked) + require.Eventually(t, func() bool { + refreshed := svc.runtimeSnapshot.Load() + return refreshed != nil && refreshed.loadedAt.After(expired.loadedAt) + }, time.Second, time.Millisecond) + + refreshed := svc.runtimeSnapshot.Load() + require.Same(t, current.config, refreshed.config) + require.Same(t, current.keywordMatcher, refreshed.keywordMatcher) + _, calls := repo.calls() + require.Equal(t, 2, calls) +} + +func TestContentModerationRuntimeSnapshotUpdateWinsOverInFlightRefresh(t *testing.T) { + repo := &contentModerationRuntimeSettingRepo{values: map[string]string{ + SettingKeyRiskControlEnabled: "true", + SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "old-keyword"), + }} + svc := runtimeCacheTestService(repo, time.Minute) + + decision, err := svc.Check(context.Background(), runtimeCacheTestInput("old-keyword")) + require.NoError(t, err) + require.True(t, decision.Blocked) + + current := svc.runtimeSnapshot.Load() + require.NotNil(t, current) + expired := *current + expired.loadedAt = time.Now().Add(-2 * time.Minute) + svc.runtimeSnapshot.Store(&expired) + + refreshStarted := make(chan struct{}, 1) + releaseRefresh := make(chan struct{}) + repo.blockNextMultiple(refreshStarted, releaseRefresh) + decision, err = svc.Check(context.Background(), runtimeCacheTestInput("clean prompt")) + require.NoError(t, err) + require.True(t, decision.Allowed) + require.Eventually(t, func() bool { + select { + case <-refreshStarted: + return true + default: + return false + } + }, time.Second, time.Millisecond) + + updateDone := make(chan error, 1) + go func() { + keywords := []string{"new-keyword"} + _, updateErr := svc.UpdateConfig(context.Background(), UpdateContentModerationConfigInput{ + BlockedKeywords: &keywords, + }) + updateDone <- updateErr + }() + select { + case updateErr := <-updateDone: + require.NoError(t, updateErr) + t.Fatal("configuration update completed before the in-flight refresh released its lock") + case <-time.After(10 * time.Millisecond): + } + + close(releaseRefresh) + require.NoError(t, <-updateDone) + decision, err = svc.Check(context.Background(), runtimeCacheTestInput("new-keyword")) + require.NoError(t, err) + require.True(t, decision.Blocked) + decision, err = svc.Check(context.Background(), runtimeCacheTestInput("old-keyword")) + require.NoError(t, err) + require.True(t, decision.Allowed) +} + +func TestContentModerationRuntimeSnapshotConcurrentReadAndReplace(t *testing.T) { + repo := &contentModerationRuntimeSettingRepo{values: map[string]string{ + SettingKeyRiskControlEnabled: "true", + SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "blocked-0"), + }} + svc := runtimeCacheTestService(repo, time.Hour) + _, err := svc.Check(context.Background(), runtimeCacheTestInput("clean prompt")) + require.NoError(t, err) + + var wg sync.WaitGroup + errs := make(chan error, 8) + for range 8 { + wg.Add(1) + go func() { + defer wg.Done() + for range 100 { + decision, checkErr := svc.Check(context.Background(), runtimeCacheTestInput("clean prompt")) + if checkErr != nil { + errs <- checkErr + return + } + if decision == nil || !decision.Allowed { + errs <- errors.New("unexpected moderation decision") + return + } + } + }() + } + for i := 1; i <= 20; i++ { + keywords := []string{"blocked-" + time.Duration(i).String()} + _, err := svc.UpdateConfig(context.Background(), UpdateContentModerationConfigInput{ + BlockedKeywords: &keywords, + }) + require.NoError(t, err) + } + wg.Wait() + close(errs) + for err := range errs { + require.NoError(t, err) + } +}