im_server/server/moderation/service.go

package moderation

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"strings"
	"sync"
	"time"

	"github.com/tinode/chat/server/logs"
	"github.com/tinode/chat/server/moderation/aliyun"
	"github.com/tinode/chat/server/store"
	"github.com/tinode/chat/server/store/types"
)

type circuitState struct {
	sync.Mutex
	failures  int
	openUntil time.Time
}

var aliyunCircuit circuitState

func Review(ctx context.Context, req Request) (*Result, error) {
	text := strings.TrimSpace(req.Text)
	result := &Result{RequestID: moderationRequestID(), Action: ActionAllow, RiskLevel: "none"}
	if text == "" || strings.TrimSpace(req.TenantCode) == "" {
		return result, nil
	}
	policy, err := store.ModerationConfig.Policy(req.TenantCode)
	if err != nil {
		return failureResult(result, nil, err)
	}
	if policy == nil || !policy.Enabled || !policy.Scenes[req.Scene] {
		return result, nil
	}
	defer func() { recordDecision(req, text, result) }()
	result.BlockMessage = policy.BlockMessage
	if policy.LocalWordsEnabled {
		words, wordsErr := store.ModerationConfig.Words(req.TenantCode)
		if wordsErr != nil {
			if logs.Warn != nil {
				logs.Warn.Printf("moderation local words unavailable: tenant=%s scene=%s err=%v", req.TenantCode, req.Scene, wordsErr)
			}
		} else if localResult := matchLocalWords(text, req.Scene, words); localResult != nil {
			result.Provider, result.RiskLevel, result.RiskLabels = "local", localResult.RiskLevel, localResult.RiskLabels
			result.Action = actionForRisk(policy, localResult.RiskLevel)
			if localResult.Summary == ActionBlock {
				result.Action = ActionBlock
			}
			if result.Action == ActionBlock {
				result.Err = ErrRejected
				return result, ErrRejected
			}
		}
	}
	if !policy.AliyunEnabled {
		return result, nil
	}
	providerConfig, err := store.ModerationConfig.Provider()
	if err != nil || providerConfig == nil || !providerConfig.Enabled {
		return failureResult(result, policy, firstError(err, ErrUnavailable))
	}
	if circuitOpen() {
		return failureResult(result, policy, ErrUnavailable)
	}
	provider, err := aliyun.NewTextClient(providerConfig)
	if err != nil {
		recordFailure()
		return failureResult(result, policy, err)
	}
	service := providerConfig.ChatService
	if req.Scene == SceneUserNickname || req.Scene == SceneUserSignature || req.Scene == SceneGroupName || req.Scene == SceneGroupNickname {
		service = providerConfig.NicknameService
	}
	providerResult, err := provider.Review(ctx, service, text, result.RequestID)
	if err != nil {
		recordFailure()
		return failureResult(result, policy, err)
	}
	recordSuccess()
	result.Provider, result.ProviderRequestID, result.RiskLevel, result.RiskLabels, result.Latency = "aliyun_green", providerResult.ProviderRequestID, providerResult.RiskLevel, providerResult.RiskLabels, providerResult.Latency
	result.Action = actionForRisk(policy, result.RiskLevel)
	if result.Action == ActionBlock {
		result.Err = ErrRejected
		return result, ErrRejected
	}
	return result, nil
}

func failureResult(result *Result, policy *types.ModerationPolicy, err error) (*Result, error) {
	result.Err = err
	failureMode := "allow"
	if policy != nil && policy.FailureMode == "block" {
		failureMode = "block"
		result.BlockMessage = policy.BlockMessage
	}
	if failureMode == "block" {
		result.Action = ActionBlock
		return result, ErrUnavailable
	}
	result.Action = ActionAuditOnly
	return result, nil
}

func actionForRisk(policy *types.ModerationPolicy, risk string) string {
	if policy != nil {
		if action := normalizedAction(policy.RiskActions[normalizedRisk(risk)]); action != "" {
			return action
		}
	}
	return ActionAllow
}
func firstError(err, fallback error) error {
	if err != nil {
		return err
	}
	return fallback
}
func moderationRequestID() string {
	var raw [12]byte
	if _, err := rand.Read(raw[:]); err != nil {
		return hex.EncodeToString([]byte(time.Now().Format(time.RFC3339Nano)))
	}
	return hex.EncodeToString(raw[:])
}
func circuitOpen() bool {
	aliyunCircuit.Lock()
	defer aliyunCircuit.Unlock()
	return time.Now().Before(aliyunCircuit.openUntil)
}
func recordFailure() {
	aliyunCircuit.Lock()
	defer aliyunCircuit.Unlock()
	aliyunCircuit.failures++
	if aliyunCircuit.failures >= 5 {
		aliyunCircuit.openUntil = time.Now().Add(30 * time.Second)
		aliyunCircuit.failures = 0
	}
}
func recordSuccess() {
	aliyunCircuit.Lock()
	aliyunCircuit.failures = 0
	aliyunCircuit.openUntil = time.Time{}
	aliyunCircuit.Unlock()
}
func logDecision(req Request, result *Result) {
	if logs.Info != nil {
		logs.Info.Printf("content moderation: request=%s tenant=%s scene=%s actor=%s provider=%s risk=%s action=%s latency_ms=%d", result.RequestID, req.TenantCode, req.Scene, req.ActorID.UserId(), result.Provider, result.RiskLevel, result.Action, result.Latency.Milliseconds())
	}
}

func recordDecision(req Request, text string, result *Result) {
	logDecision(req, result)
	sum := sha256.Sum256([]byte(text))
	errorCode := ""
	if result.Err != nil {
		errorCode = "PROVIDER_ERROR"
	}
	if err := store.SecurityAudit.RecordContentModeration(req.TenantID, &types.ContentModerationLog{
		RequestID: result.RequestID, Scene: req.Scene, Topic: req.Topic, ActorID: req.ActorID,
		Provider: result.Provider, ProviderRequestID: result.ProviderRequestID, RiskLevel: result.RiskLevel,
		RiskLabels: result.RiskLabels, Action: result.Action, ContentHash: hex.EncodeToString(sum[:]),
		LatencyMS: result.Latency.Milliseconds(), ErrorCode: errorCode, CreatedAt: time.Now().UTC().Round(time.Millisecond),
	}); err != nil {
		if logs.Warn != nil {
			logs.Warn.Printf("content moderation audit write failed: request=%s tenant=%s err=%v", result.RequestID, req.TenantCode, err)
		}
	}
}