2585 lines
76 KiB
Go
2585 lines
76 KiB
Go
|
package compliance
|
|||
|
|
|
|||
|
|
import (
|
|||
|
|
"context"
|
|||
|
|
"crypto/sha256"
|
|||
|
|
"encoding/hex"
|
|||
|
|
"encoding/json"
|
|||
|
|
"errors"
|
|||
|
|
"fmt"
|
|||
|
|
"regexp"
|
|||
|
|
"sort"
|
|||
|
|
"strings"
|
|||
|
|
"sync"
|
|||
|
|
"time"
|
|||
|
|
"unicode/utf8"
|
|||
|
|
|
|||
|
|
"github.com/google/uuid"
|
|||
|
|
"github.com/jackc/pgx/v5"
|
|||
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|||
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|||
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|||
|
|
"go.uber.org/zap"
|
|||
|
|
|
|||
|
|
sharedcompliance "github.com/geo-platform/tenant-api/internal/shared/compliance"
|
|||
|
|
"github.com/geo-platform/tenant-api/internal/shared/config"
|
|||
|
|
sharedllm "github.com/geo-platform/tenant-api/internal/shared/llm"
|
|||
|
|
"github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq"
|
|||
|
|
"github.com/geo-platform/tenant-api/internal/shared/response"
|
|||
|
|
tenantrepo "github.com/geo-platform/tenant-api/internal/tenant/repository"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
const (
|
|||
|
|
triggerEditorManual = "editor_manual"
|
|||
|
|
triggerPublishGate = "publish_gate"
|
|||
|
|
triggerAPI = "api"
|
|||
|
|
triggerSchedulerRecheck = "scheduler_recheck"
|
|||
|
|
triggerManualReviewBypass = "manual_review_bypass"
|
|||
|
|
globalPolicyCacheKey = int64(0)
|
|||
|
|
|
|||
|
|
sourceDict = "dict"
|
|||
|
|
sourceRegex = "regex"
|
|||
|
|
|
|||
|
|
aiPointsQuotaType = "ai_points"
|
|||
|
|
aiUsageTypeComplianceLLMJudge = "compliance_llm_judge"
|
|||
|
|
aiResourceTypeComplianceReview = "compliance_review"
|
|||
|
|
|
|||
|
|
maxStoredViolations = 200
|
|||
|
|
maxMatchedTextRunes = 80
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
type Service struct {
|
|||
|
|
pool *pgxpool.Pool
|
|||
|
|
mq *rabbitmq.Client
|
|||
|
|
cfg config.Provider
|
|||
|
|
llm sharedllm.Client
|
|||
|
|
logger *zap.Logger
|
|||
|
|
policyMu sync.Mutex
|
|||
|
|
policies map[int64]cachedPolicy
|
|||
|
|
|
|||
|
|
reviewCache sync.Map
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type cachedPolicy struct {
|
|||
|
|
policy *policySnapshot
|
|||
|
|
expires time.Time
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func NewService(pool *pgxpool.Pool, cfg config.Provider, logger *zap.Logger) *Service {
|
|||
|
|
return &Service{pool: pool, cfg: cfg, logger: logger, policies: make(map[int64]cachedPolicy)}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) WithRabbitMQ(mq *rabbitmq.Client) *Service {
|
|||
|
|
if s != nil {
|
|||
|
|
s.mq = mq
|
|||
|
|
}
|
|||
|
|
return s
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) WithLLM(client sharedllm.Client) *Service {
|
|||
|
|
if s != nil {
|
|||
|
|
s.llm = client
|
|||
|
|
}
|
|||
|
|
return s
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type GateForPublishRequest struct {
|
|||
|
|
TenantID int64
|
|||
|
|
ArticleID int64
|
|||
|
|
ArticleVersionID int64
|
|||
|
|
ActorID int64
|
|||
|
|
TargetPlatforms []string
|
|||
|
|
AckRecordID *int64
|
|||
|
|
TriggerSource string
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type CheckArticleRequest struct {
|
|||
|
|
TenantID int64
|
|||
|
|
ArticleID int64
|
|||
|
|
ArticleVersionID *int64
|
|||
|
|
ActorID int64
|
|||
|
|
Title *string
|
|||
|
|
Plaintext *string
|
|||
|
|
TargetPlatforms []string
|
|||
|
|
TriggerSource string
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type CheckPlainRequest struct {
|
|||
|
|
TenantID int64
|
|||
|
|
ActorID int64
|
|||
|
|
Title string
|
|||
|
|
Plaintext string
|
|||
|
|
TargetPlatforms []string
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type ConsumeAckQuery struct {
|
|||
|
|
TenantID int64
|
|||
|
|
ArticleID int64
|
|||
|
|
ArticleVersionID int64
|
|||
|
|
CheckRecordID int64
|
|||
|
|
ContentHash string
|
|||
|
|
PolicyFingerprint string
|
|||
|
|
ConsumedByJobID uuid.UUID
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type CreateManualReviewRequest struct {
|
|||
|
|
TenantID int64
|
|||
|
|
ArticleID int64
|
|||
|
|
ActorID int64
|
|||
|
|
CheckRecordID int64
|
|||
|
|
Note string
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type CancelManualReviewRequest struct {
|
|||
|
|
TenantID int64
|
|||
|
|
ArticleID int64
|
|||
|
|
ActorID int64
|
|||
|
|
ReviewID int64
|
|||
|
|
Reason string
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type ReviewJobMessage struct {
|
|||
|
|
JobID int64 `json:"job_id"`
|
|||
|
|
MessageID string `json:"message_id"`
|
|||
|
|
TenantID int64 `json:"tenant_id"`
|
|||
|
|
ArticleID int64 `json:"article_id"`
|
|||
|
|
ArticleVersionID int64 `json:"article_version_id"`
|
|||
|
|
CheckRecordID int64 `json:"check_record_id"`
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type GateOutcome struct {
|
|||
|
|
Decision sharedcompliance.GateDecision
|
|||
|
|
Result *sharedcompliance.CheckResult
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type articleVersionSnapshot struct {
|
|||
|
|
ArticleID int64
|
|||
|
|
ArticleVersionID int64
|
|||
|
|
Title string
|
|||
|
|
Plaintext string
|
|||
|
|
ManualReviewStatus sharedcompliance.ManualReviewStatus
|
|||
|
|
ManualReviewID *int64
|
|||
|
|
ManualReviewReason *string
|
|||
|
|
ManualReviewDecidedAt *time.Time
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type policySnapshot struct {
|
|||
|
|
LoadedAt time.Time
|
|||
|
|
MasterEnabled bool
|
|||
|
|
EnforcementMode string
|
|||
|
|
LLMJudgeEnabled bool
|
|||
|
|
Revision int64
|
|||
|
|
DictionaryVersionSum int64
|
|||
|
|
EnabledDictionaries []string
|
|||
|
|
DictVersions map[string]int
|
|||
|
|
Terms []termRule
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type termRule struct {
|
|||
|
|
DictionaryID int64
|
|||
|
|
DictionaryCode string
|
|||
|
|
DictionaryName string
|
|||
|
|
DictionaryVersion int
|
|||
|
|
PlatformScope string
|
|||
|
|
ApplicablePlatforms []string
|
|||
|
|
MatchType string
|
|||
|
|
Pattern string
|
|||
|
|
Level string
|
|||
|
|
Hint *string
|
|||
|
|
ReferenceLaw *string
|
|||
|
|
compiledRegex *regexp.Regexp
|
|||
|
|
compiledRegexPattern string
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type checkRecordRow struct {
|
|||
|
|
ID int64
|
|||
|
|
TenantID int64
|
|||
|
|
ArticleID int64
|
|||
|
|
ArticleVersionID int64
|
|||
|
|
TargetPlatforms []string
|
|||
|
|
ContentHash string
|
|||
|
|
PolicyFingerprint string
|
|||
|
|
EnforcementMode string
|
|||
|
|
HighestLevel *string
|
|||
|
|
Passed bool
|
|||
|
|
HitCount int
|
|||
|
|
StoredHitCount int
|
|||
|
|
Truncated bool
|
|||
|
|
ManualReviewStatus sharedcompliance.ManualReviewStatus
|
|||
|
|
ManualReviewID *int64
|
|||
|
|
CheckedAt time.Time
|
|||
|
|
DurationMS int
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) RuntimeStatus(ctx context.Context, tenantID int64) (*sharedcompliance.RuntimeStatus, error) {
|
|||
|
|
cfg := s.currentComplianceConfig()
|
|||
|
|
if !cfg.Enabled {
|
|||
|
|
return &sharedcompliance.RuntimeStatus{
|
|||
|
|
Enabled: false,
|
|||
|
|
ConfigEnabled: false,
|
|||
|
|
MasterEnabled: false,
|
|||
|
|
EnforcementMode: "disabled",
|
|||
|
|
LLMJudgeEnabled: false,
|
|||
|
|
SnapshotReloadInterval: cfg.SnapshotReloadInterval.String(),
|
|||
|
|
PublishGateTimeout: cfg.PublishGateTimeout.String(),
|
|||
|
|
}, nil
|
|||
|
|
}
|
|||
|
|
policy, err := s.resolvePolicy(ctx, tenantID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
return &sharedcompliance.RuntimeStatus{
|
|||
|
|
Enabled: policy.MasterEnabled && policy.EnforcementMode != "disabled",
|
|||
|
|
ConfigEnabled: true,
|
|||
|
|
MasterEnabled: policy.MasterEnabled,
|
|||
|
|
EnforcementMode: policy.EnforcementMode,
|
|||
|
|
LLMJudgeEnabled: policy.LLMJudgeEnabled,
|
|||
|
|
SnapshotReloadInterval: cfg.SnapshotReloadInterval.String(),
|
|||
|
|
PublishGateTimeout: cfg.PublishGateTimeout.String(),
|
|||
|
|
DictionaryVersion: policy.DictionaryVersionSum,
|
|||
|
|
EnabledDictionaries: append([]string(nil), policy.EnabledDictionaries...),
|
|||
|
|
}, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) CheckPlain(ctx context.Context, req CheckPlainRequest) (*sharedcompliance.CheckResult, error) {
|
|||
|
|
cfg := s.currentComplianceConfig()
|
|||
|
|
if !cfg.Enabled {
|
|||
|
|
return disabledCheckResult(req.TargetPlatforms), nil
|
|||
|
|
}
|
|||
|
|
policy, err := s.resolvePolicy(ctx, req.TenantID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
return s.runAndPersist(ctx, runInput{
|
|||
|
|
TenantID: req.TenantID,
|
|||
|
|
ActorID: req.ActorID,
|
|||
|
|
Title: req.Title,
|
|||
|
|
Plaintext: req.Plaintext,
|
|||
|
|
TargetPlatforms: req.TargetPlatforms,
|
|||
|
|
TriggerSource: triggerAPI,
|
|||
|
|
Policy: policy,
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) CheckArticle(ctx context.Context, req CheckArticleRequest) (*sharedcompliance.CheckResult, error) {
|
|||
|
|
cfg := s.currentComplianceConfig()
|
|||
|
|
targets := normalizePlatforms(req.TargetPlatforms)
|
|||
|
|
trigger := normalizeArticleCheckTrigger(req.TriggerSource)
|
|||
|
|
if !cfg.Enabled && trigger != triggerEditorManual {
|
|||
|
|
return disabledCheckResult(targets), nil
|
|||
|
|
}
|
|||
|
|
versionID, err := s.ResolvePublishableVersion(ctx, req.TenantID, req.ArticleID, req.ArticleVersionID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
snapshot, err := s.loadArticleVersionSnapshot(ctx, req.TenantID, req.ArticleID, versionID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
if snapshot.ManualReviewStatus == sharedcompliance.ManualReviewStatusPending ||
|
|||
|
|
snapshot.ManualReviewStatus == sharedcompliance.ManualReviewStatusApproved ||
|
|||
|
|
snapshot.ManualReviewStatus == sharedcompliance.ManualReviewStatusRejected {
|
|||
|
|
if trigger != triggerEditorManual {
|
|||
|
|
return manualReviewOnlyResult(snapshot, targets), nil
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
policy, err := s.resolvePolicy(ctx, req.TenantID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
title := snapshot.Title
|
|||
|
|
if req.Title != nil {
|
|||
|
|
title = strings.TrimSpace(*req.Title)
|
|||
|
|
}
|
|||
|
|
plaintext := snapshot.Plaintext
|
|||
|
|
if req.Plaintext != nil {
|
|||
|
|
plaintext = strings.TrimSpace(*req.Plaintext)
|
|||
|
|
}
|
|||
|
|
if trigger == triggerEditorManual {
|
|||
|
|
return s.runPreview(ctx, runInput{
|
|||
|
|
TenantID: req.TenantID,
|
|||
|
|
ArticleID: req.ArticleID,
|
|||
|
|
ArticleVersionID: versionID,
|
|||
|
|
ActorID: req.ActorID,
|
|||
|
|
Title: title,
|
|||
|
|
Plaintext: plaintext,
|
|||
|
|
TargetPlatforms: targets,
|
|||
|
|
TriggerSource: trigger,
|
|||
|
|
Policy: editorPreviewPolicy(policy),
|
|||
|
|
ManualStatus: snapshot.ManualReviewStatus,
|
|||
|
|
ManualReviewID: snapshot.ManualReviewID,
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
return s.runAndPersist(ctx, runInput{
|
|||
|
|
TenantID: req.TenantID,
|
|||
|
|
ArticleID: req.ArticleID,
|
|||
|
|
ArticleVersionID: versionID,
|
|||
|
|
ActorID: req.ActorID,
|
|||
|
|
Title: title,
|
|||
|
|
Plaintext: plaintext,
|
|||
|
|
TargetPlatforms: targets,
|
|||
|
|
TriggerSource: trigger,
|
|||
|
|
Policy: policy,
|
|||
|
|
ManualStatus: snapshot.ManualReviewStatus,
|
|||
|
|
ManualReviewID: snapshot.ManualReviewID,
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) GateForPublish(ctx context.Context, req GateForPublishRequest) (*GateOutcome, error) {
|
|||
|
|
cfg := s.currentComplianceConfig()
|
|||
|
|
if !cfg.Enabled {
|
|||
|
|
result := disabledCheckResult(req.TargetPlatforms)
|
|||
|
|
result.Decision = sharedcompliance.GateDecisionPass
|
|||
|
|
result.Passed = true
|
|||
|
|
return &GateOutcome{Decision: sharedcompliance.GateDecisionPass, Result: result}, nil
|
|||
|
|
}
|
|||
|
|
req.TargetPlatforms = normalizePlatforms(req.TargetPlatforms)
|
|||
|
|
trigger := strings.TrimSpace(req.TriggerSource)
|
|||
|
|
if trigger == "" {
|
|||
|
|
trigger = triggerPublishGate
|
|||
|
|
}
|
|||
|
|
req.TriggerSource = trigger
|
|||
|
|
|
|||
|
|
snapshot, err := s.loadArticleVersionSnapshot(ctx, req.TenantID, req.ArticleID, req.ArticleVersionID)
|
|||
|
|
if err != nil {
|
|||
|
|
if cached, ok := s.cachedManualReview(req.ArticleVersionID); ok && cached.ManualReviewStatus != sharedcompliance.ManualReviewStatusNone {
|
|||
|
|
result := manualReviewOnlyResult(cached, req.TargetPlatforms)
|
|||
|
|
return &GateOutcome{Decision: result.Decision, Result: result}, manualReviewGateError(cached)
|
|||
|
|
}
|
|||
|
|
return nil, response.ErrServiceUnavailable(41004, "compliance_engine_unavailable", "manual review status is unavailable")
|
|||
|
|
}
|
|||
|
|
s.cacheManualReview(snapshot)
|
|||
|
|
|
|||
|
|
switch snapshot.ManualReviewStatus {
|
|||
|
|
case sharedcompliance.ManualReviewStatusPending:
|
|||
|
|
result := manualReviewOnlyResult(snapshot, req.TargetPlatforms)
|
|||
|
|
return &GateOutcome{Decision: sharedcompliance.GateDecisionBlock, Result: result}, response.ErrConflict(41007, "compliance_manual_review_pending", "current article version is waiting for manual review")
|
|||
|
|
case sharedcompliance.ManualReviewStatusApproved:
|
|||
|
|
result, err := s.createManualReviewBypassRecord(ctx, req, snapshot)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
return &GateOutcome{Decision: sharedcompliance.GateDecisionPass, Result: result}, nil
|
|||
|
|
case sharedcompliance.ManualReviewStatusRejected:
|
|||
|
|
result := manualReviewOnlyResult(snapshot, req.TargetPlatforms)
|
|||
|
|
return &GateOutcome{Decision: sharedcompliance.GateDecisionBlock, Result: result}, response.ErrConflict(41006, "compliance_manual_review_rejected", "current article version was rejected by manual review")
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
policy, err := s.resolvePolicy(ctx, req.TenantID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
if !policy.MasterEnabled || policy.EnforcementMode == "disabled" {
|
|||
|
|
result := disabledCheckResult(req.TargetPlatforms)
|
|||
|
|
result.EnforcementMode = policy.EnforcementMode
|
|||
|
|
result.Passed = true
|
|||
|
|
result.Decision = sharedcompliance.GateDecisionPass
|
|||
|
|
return &GateOutcome{Decision: sharedcompliance.GateDecisionPass, Result: result}, nil
|
|||
|
|
}
|
|||
|
|
if req.AckRecordID != nil {
|
|||
|
|
result, err := s.validateAckForPublish(ctx, req, policy)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
return &GateOutcome{Decision: sharedcompliance.GateDecisionPass, Result: result}, nil
|
|||
|
|
}
|
|||
|
|
result, err := s.runAndPersist(ctx, runInput{
|
|||
|
|
TenantID: req.TenantID,
|
|||
|
|
ArticleID: req.ArticleID,
|
|||
|
|
ArticleVersionID: req.ArticleVersionID,
|
|||
|
|
ActorID: req.ActorID,
|
|||
|
|
Title: snapshot.Title,
|
|||
|
|
Plaintext: snapshot.Plaintext,
|
|||
|
|
TargetPlatforms: req.TargetPlatforms,
|
|||
|
|
TriggerSource: trigger,
|
|||
|
|
Policy: policy,
|
|||
|
|
ManualStatus: snapshot.ManualReviewStatus,
|
|||
|
|
ManualReviewID: snapshot.ManualReviewID,
|
|||
|
|
})
|
|||
|
|
if err != nil {
|
|||
|
|
if policy.EnforcementMode == "mandatory" {
|
|||
|
|
return nil, response.ErrServiceUnavailable(41004, "compliance_engine_unavailable", "compliance gate failed closed")
|
|||
|
|
}
|
|||
|
|
return &GateOutcome{Decision: sharedcompliance.GateDecisionPass, Result: disabledCheckResult(req.TargetPlatforms)}, nil
|
|||
|
|
}
|
|||
|
|
return &GateOutcome{Decision: result.Decision, Result: result}, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) ResolvePublishableVersion(ctx context.Context, tenantID, articleID int64, requested *int64) (int64, error) {
|
|||
|
|
if requested != nil && *requested > 0 {
|
|||
|
|
var id int64
|
|||
|
|
err := s.pool.QueryRow(ctx, `
|
|||
|
|
SELECT av.id
|
|||
|
|
FROM article_versions av
|
|||
|
|
JOIN articles a ON a.id = av.article_id
|
|||
|
|
WHERE av.id = $1
|
|||
|
|
AND av.article_id = $2
|
|||
|
|
AND a.tenant_id = $3
|
|||
|
|
AND a.deleted_at IS NULL
|
|||
|
|
`, *requested, articleID, tenantID).Scan(&id)
|
|||
|
|
if err != nil {
|
|||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|||
|
|
return 0, response.ErrBadRequest(41005, "invalid_article_version", "article_version_id does not belong to this article")
|
|||
|
|
}
|
|||
|
|
return 0, response.ErrInternal(50010, "query_failed", "failed to validate article version")
|
|||
|
|
}
|
|||
|
|
return id, nil
|
|||
|
|
}
|
|||
|
|
var versionValue pgtype.Int8
|
|||
|
|
err := s.pool.QueryRow(ctx, `
|
|||
|
|
SELECT current_version_id
|
|||
|
|
FROM articles
|
|||
|
|
WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL
|
|||
|
|
`, articleID, tenantID).Scan(&versionValue)
|
|||
|
|
if err != nil {
|
|||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|||
|
|
return 0, response.ErrNotFound(40411, "article_not_found", "article not found")
|
|||
|
|
}
|
|||
|
|
return 0, response.ErrInternal(50010, "query_failed", "failed to resolve current article version")
|
|||
|
|
}
|
|||
|
|
if !versionValue.Valid || versionValue.Int64 <= 0 {
|
|||
|
|
return 0, response.ErrBadRequest(41005, "invalid_article_version", "article has no publishable version")
|
|||
|
|
}
|
|||
|
|
return versionValue.Int64, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type runInput struct {
|
|||
|
|
TenantID int64
|
|||
|
|
ArticleID int64
|
|||
|
|
ArticleVersionID int64
|
|||
|
|
ActorID int64
|
|||
|
|
Title string
|
|||
|
|
Plaintext string
|
|||
|
|
TargetPlatforms []string
|
|||
|
|
TriggerSource string
|
|||
|
|
Policy *policySnapshot
|
|||
|
|
ManualStatus sharedcompliance.ManualReviewStatus
|
|||
|
|
ManualReviewID *int64
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) runAndPersist(ctx context.Context, input runInput) (*sharedcompliance.CheckResult, error) {
|
|||
|
|
checkResult, violations, err := s.runCheck(input)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
persistedResult, err := s.insertCheckRecord(ctx, input, checkRecordRow{
|
|||
|
|
TenantID: input.TenantID,
|
|||
|
|
ArticleID: input.ArticleID,
|
|||
|
|
ArticleVersionID: input.ArticleVersionID,
|
|||
|
|
TargetPlatforms: checkResult.TargetPlatforms,
|
|||
|
|
ContentHash: checkResult.ContentHash,
|
|||
|
|
PolicyFingerprint: checkResult.PolicyFingerprint,
|
|||
|
|
EnforcementMode: checkResult.EnforcementMode,
|
|||
|
|
HighestLevel: checkResult.HighestLevel,
|
|||
|
|
Passed: checkResult.Passed,
|
|||
|
|
HitCount: checkResult.HitCount,
|
|||
|
|
StoredHitCount: checkResult.StoredHitCount,
|
|||
|
|
Truncated: checkResult.Truncated,
|
|||
|
|
ManualReviewStatus: checkResult.ManualReviewStatus,
|
|||
|
|
ManualReviewID: checkResult.ManualReviewID,
|
|||
|
|
CheckedAt: checkResult.CheckedAt,
|
|||
|
|
DurationMS: checkResult.DurationMS,
|
|||
|
|
}, violations, checkResult.Decision)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
if input.Policy.LLMJudgeEnabled && input.ArticleID > 0 && input.ArticleVersionID > 0 && persistedResult.RecordID > 0 {
|
|||
|
|
s.enqueueLLMReview(ctx, input, persistedResult.RecordID)
|
|||
|
|
}
|
|||
|
|
return persistedResult, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) runPreview(ctx context.Context, input runInput) (*sharedcompliance.CheckResult, error) {
|
|||
|
|
result, _, err := s.runCheck(input)
|
|||
|
|
return result, err
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) runCheck(input runInput) (*sharedcompliance.CheckResult, []sharedcompliance.Violation, error) {
|
|||
|
|
start := time.Now()
|
|||
|
|
if input.Policy == nil {
|
|||
|
|
return nil, nil, response.ErrServiceUnavailable(41004, "compliance_engine_unavailable", "compliance policy is unavailable")
|
|||
|
|
}
|
|||
|
|
input.TargetPlatforms = normalizePlatforms(input.TargetPlatforms)
|
|||
|
|
plaintext := strings.TrimSpace(input.Plaintext)
|
|||
|
|
title := strings.TrimSpace(input.Title)
|
|||
|
|
contentHash := contentHash(title, plaintext)
|
|||
|
|
fingerprint := policyFingerprint(input.Policy, input.TargetPlatforms)
|
|||
|
|
violations := matchViolations(input.Policy, input.TenantID, input.TargetPlatforms, title+"\n"+plaintext)
|
|||
|
|
highest := highestViolationLevel(violations)
|
|||
|
|
hitCount := len(violations)
|
|||
|
|
truncated := false
|
|||
|
|
if len(violations) > maxStoredViolations {
|
|||
|
|
violations = violations[:maxStoredViolations]
|
|||
|
|
truncated = true
|
|||
|
|
}
|
|||
|
|
decision := decide(input.Policy.EnforcementMode, highest, hitCount)
|
|||
|
|
passed := decision == sharedcompliance.GateDecisionPass
|
|||
|
|
durationMS := int(time.Since(start).Milliseconds())
|
|||
|
|
if input.ManualStatus == "" {
|
|||
|
|
input.ManualStatus = sharedcompliance.ManualReviewStatusNone
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
result := &sharedcompliance.CheckResult{
|
|||
|
|
Decision: decision,
|
|||
|
|
Passed: passed,
|
|||
|
|
HighestLevel: highest,
|
|||
|
|
Violations: violations,
|
|||
|
|
HitCount: hitCount,
|
|||
|
|
StoredHitCount: len(violations),
|
|||
|
|
Truncated: truncated,
|
|||
|
|
EnforcementMode: input.Policy.EnforcementMode,
|
|||
|
|
ContentHash: contentHash,
|
|||
|
|
PolicyFingerprint: fingerprint,
|
|||
|
|
TargetPlatforms: input.TargetPlatforms,
|
|||
|
|
ManualReviewStatus: input.ManualStatus,
|
|||
|
|
ManualReviewID: input.ManualReviewID,
|
|||
|
|
CheckedAt: time.Now().UTC(),
|
|||
|
|
DurationMS: durationMS,
|
|||
|
|
}
|
|||
|
|
if input.TriggerSource == triggerEditorManual {
|
|||
|
|
result.StoredHitCount = 0
|
|||
|
|
}
|
|||
|
|
return result, violations, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) insertCheckRecord(ctx context.Context, input runInput, record checkRecordRow, violations []sharedcompliance.Violation, decision sharedcompliance.GateDecision) (*sharedcompliance.CheckResult, error) {
|
|||
|
|
dictVersions, _ := json.Marshal(recordDictVersions(input.Policy))
|
|||
|
|
var articleID any
|
|||
|
|
if record.ArticleID > 0 {
|
|||
|
|
articleID = &record.ArticleID
|
|||
|
|
}
|
|||
|
|
var articleVersionID any
|
|||
|
|
if record.ArticleVersionID > 0 {
|
|||
|
|
articleVersionID = &record.ArticleVersionID
|
|||
|
|
}
|
|||
|
|
var checkedBy any
|
|||
|
|
if input.ActorID > 0 {
|
|||
|
|
checkedBy = &input.ActorID
|
|||
|
|
}
|
|||
|
|
var highest any
|
|||
|
|
if record.HighestLevel != nil {
|
|||
|
|
highest = *record.HighestLevel
|
|||
|
|
}
|
|||
|
|
err := s.pool.QueryRow(ctx, `
|
|||
|
|
INSERT INTO compliance_check_records (
|
|||
|
|
tenant_id,
|
|||
|
|
article_id,
|
|||
|
|
article_version_id,
|
|||
|
|
target_platforms,
|
|||
|
|
content_hash,
|
|||
|
|
policy_fingerprint,
|
|||
|
|
trigger_source,
|
|||
|
|
enforcement_mode,
|
|||
|
|
highest_level,
|
|||
|
|
passed,
|
|||
|
|
hit_count,
|
|||
|
|
stored_hit_count,
|
|||
|
|
truncated,
|
|||
|
|
dict_versions,
|
|||
|
|
llm_used,
|
|||
|
|
duration_ms,
|
|||
|
|
checked_by,
|
|||
|
|
manual_review_status,
|
|||
|
|
manual_review_id
|
|||
|
|
)
|
|||
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,FALSE,$15,$16,$17,$18)
|
|||
|
|
RETURNING id, checked_at
|
|||
|
|
`,
|
|||
|
|
record.TenantID,
|
|||
|
|
articleID,
|
|||
|
|
articleVersionID,
|
|||
|
|
record.TargetPlatforms,
|
|||
|
|
record.ContentHash,
|
|||
|
|
record.PolicyFingerprint,
|
|||
|
|
input.TriggerSource,
|
|||
|
|
record.EnforcementMode,
|
|||
|
|
highest,
|
|||
|
|
record.Passed,
|
|||
|
|
record.HitCount,
|
|||
|
|
record.StoredHitCount,
|
|||
|
|
record.Truncated,
|
|||
|
|
dictVersions,
|
|||
|
|
record.DurationMS,
|
|||
|
|
checkedBy,
|
|||
|
|
string(record.ManualReviewStatus),
|
|||
|
|
int64PtrAny(record.ManualReviewID),
|
|||
|
|
).Scan(&record.ID, &record.CheckedAt)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, response.ErrInternal(51001, "compliance_check_record_create_failed", "failed to create compliance check record")
|
|||
|
|
}
|
|||
|
|
for i := range violations {
|
|||
|
|
violations[i].RecordID = record.ID
|
|||
|
|
id, err := s.insertViolation(ctx, record.ID, record.TenantID, violations[i])
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
violations[i].ID = id
|
|||
|
|
}
|
|||
|
|
return checkResultFromRecord(record, violations, decision), nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) insertViolation(ctx context.Context, recordID, tenantID int64, v sharedcompliance.Violation) (int64, error) {
|
|||
|
|
var id int64
|
|||
|
|
err := s.pool.QueryRow(ctx, `
|
|||
|
|
INSERT INTO compliance_violations (
|
|||
|
|
record_id,
|
|||
|
|
tenant_id,
|
|||
|
|
platform_code,
|
|||
|
|
source,
|
|||
|
|
dictionary_code,
|
|||
|
|
dictionary_name,
|
|||
|
|
term_pattern,
|
|||
|
|
matched_text,
|
|||
|
|
start_offset,
|
|||
|
|
end_offset,
|
|||
|
|
dom_path,
|
|||
|
|
level,
|
|||
|
|
hint,
|
|||
|
|
reference_law
|
|||
|
|
)
|
|||
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)
|
|||
|
|
RETURNING id
|
|||
|
|
`,
|
|||
|
|
recordID,
|
|||
|
|
tenantID,
|
|||
|
|
v.PlatformCode,
|
|||
|
|
v.Source,
|
|||
|
|
v.DictionaryCode,
|
|||
|
|
v.DictionaryName,
|
|||
|
|
v.TermPattern,
|
|||
|
|
truncateRunes(v.MatchedText, maxMatchedTextRunes),
|
|||
|
|
v.StartOffset,
|
|||
|
|
v.EndOffset,
|
|||
|
|
v.DomPath,
|
|||
|
|
v.Level,
|
|||
|
|
v.Hint,
|
|||
|
|
v.ReferenceLaw,
|
|||
|
|
).Scan(&id)
|
|||
|
|
if err != nil {
|
|||
|
|
return 0, response.ErrInternal(51002, "compliance_violation_create_failed", "failed to create compliance violation")
|
|||
|
|
}
|
|||
|
|
return id, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) CreateAck(ctx context.Context, tenantID, checkRecordID, actorID int64, acknowledgedViolationIDs []int64) (*sharedcompliance.AckRecord, error) {
|
|||
|
|
record, err := s.loadCheckRecord(ctx, tenantID, checkRecordID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
if record.ArticleID <= 0 || record.ArticleVersionID <= 0 {
|
|||
|
|
return nil, response.ErrBadRequest(41003, "compliance_ack_mismatch", "check record is not bound to an article version")
|
|||
|
|
}
|
|||
|
|
violations, err := s.loadViolations(ctx, tenantID, checkRecordID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
required := make([]int64, 0, len(violations))
|
|||
|
|
for _, violation := range violations {
|
|||
|
|
if record.EnforcementMode == "mandatory" && violation.Level == "block" {
|
|||
|
|
return nil, response.ErrConflict(41001, "compliance_blocked", "blocked violations cannot be acknowledged")
|
|||
|
|
}
|
|||
|
|
required = append(required, violation.ID)
|
|||
|
|
}
|
|||
|
|
if !sameInt64Set(required, acknowledgedViolationIDs) {
|
|||
|
|
return nil, response.ErrBadRequest(41003, "compliance_ack_mismatch", "acknowledged_violation_ids must cover every acknowledged violation")
|
|||
|
|
}
|
|||
|
|
if len(required) == 0 {
|
|||
|
|
return nil, response.ErrBadRequest(41003, "compliance_ack_mismatch", "there are no violations to acknowledge")
|
|||
|
|
}
|
|||
|
|
tx, err := s.pool.Begin(ctx)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, response.ErrInternal(51003, "compliance_ack_create_failed", "failed to begin compliance ack transaction")
|
|||
|
|
}
|
|||
|
|
defer rollbackTx(tx)
|
|||
|
|
expiresAt := time.Now().UTC().Add(24 * time.Hour)
|
|||
|
|
var ack sharedcompliance.AckRecord
|
|||
|
|
err = tx.QueryRow(ctx, `
|
|||
|
|
INSERT INTO compliance_ack_records (
|
|||
|
|
tenant_id,
|
|||
|
|
article_id,
|
|||
|
|
article_version_id,
|
|||
|
|
check_record_id,
|
|||
|
|
content_hash,
|
|||
|
|
policy_fingerprint,
|
|||
|
|
acknowledged_by,
|
|||
|
|
ack_violation_ids,
|
|||
|
|
expires_at
|
|||
|
|
)
|
|||
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
|
|||
|
|
RETURNING id, check_record_id, article_id, article_version_id, content_hash, policy_fingerprint, expires_at, created_at
|
|||
|
|
`,
|
|||
|
|
tenantID,
|
|||
|
|
record.ArticleID,
|
|||
|
|
record.ArticleVersionID,
|
|||
|
|
record.ID,
|
|||
|
|
record.ContentHash,
|
|||
|
|
record.PolicyFingerprint,
|
|||
|
|
actorID,
|
|||
|
|
required,
|
|||
|
|
expiresAt,
|
|||
|
|
).Scan(&ack.ID, &ack.CheckRecordID, &ack.ArticleID, &ack.ArticleVersionID, &ack.ContentHash, &ack.PolicyFingerprint, &ack.ExpiresAt, &ack.CreatedAt)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, response.ErrInternal(51003, "compliance_ack_create_failed", "failed to create compliance ack")
|
|||
|
|
}
|
|||
|
|
if _, err := tx.Exec(ctx, `
|
|||
|
|
UPDATE compliance_violations
|
|||
|
|
SET acknowledged_in_ack_id = $1
|
|||
|
|
WHERE record_id = $2 AND tenant_id = $3 AND id = ANY($4)
|
|||
|
|
`, ack.ID, record.ID, tenantID, required); err != nil {
|
|||
|
|
return nil, response.ErrInternal(51003, "compliance_ack_create_failed", "failed to mark acknowledged violations")
|
|||
|
|
}
|
|||
|
|
if err := tx.Commit(ctx); err != nil {
|
|||
|
|
return nil, response.ErrInternal(51003, "compliance_ack_create_failed", "failed to commit compliance ack")
|
|||
|
|
}
|
|||
|
|
return &ack, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) ConsumeAckTx(ctx context.Context, tx pgx.Tx, ackRecordID int64, q ConsumeAckQuery) (bool, error) {
|
|||
|
|
if tx == nil {
|
|||
|
|
return false, errors.New("tx is nil")
|
|||
|
|
}
|
|||
|
|
tag, err := tx.Exec(ctx, `
|
|||
|
|
UPDATE compliance_ack_records
|
|||
|
|
SET consumed_at = NOW(), consumed_by_job_id = $1
|
|||
|
|
WHERE id = $2
|
|||
|
|
AND tenant_id = $3
|
|||
|
|
AND article_id = $4
|
|||
|
|
AND article_version_id = $5
|
|||
|
|
AND check_record_id = $6
|
|||
|
|
AND content_hash = $7
|
|||
|
|
AND policy_fingerprint = $8
|
|||
|
|
AND consumed_at IS NULL
|
|||
|
|
AND expires_at > NOW()
|
|||
|
|
`,
|
|||
|
|
q.ConsumedByJobID,
|
|||
|
|
ackRecordID,
|
|||
|
|
q.TenantID,
|
|||
|
|
q.ArticleID,
|
|||
|
|
q.ArticleVersionID,
|
|||
|
|
q.CheckRecordID,
|
|||
|
|
q.ContentHash,
|
|||
|
|
q.PolicyFingerprint,
|
|||
|
|
)
|
|||
|
|
if err != nil {
|
|||
|
|
return false, err
|
|||
|
|
}
|
|||
|
|
return tag.RowsAffected() == 1, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) validateAckForPublish(ctx context.Context, req GateForPublishRequest, policy *policySnapshot) (*sharedcompliance.CheckResult, error) {
|
|||
|
|
var recordID int64
|
|||
|
|
err := s.pool.QueryRow(ctx, `
|
|||
|
|
SELECT check_record_id
|
|||
|
|
FROM compliance_ack_records
|
|||
|
|
WHERE id = $1
|
|||
|
|
AND tenant_id = $2
|
|||
|
|
AND article_id = $3
|
|||
|
|
AND article_version_id = $4
|
|||
|
|
AND consumed_at IS NULL
|
|||
|
|
AND expires_at > NOW()
|
|||
|
|
`, *req.AckRecordID, req.TenantID, req.ArticleID, req.ArticleVersionID).Scan(&recordID)
|
|||
|
|
if err != nil {
|
|||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|||
|
|
return nil, response.ErrBadRequest(41003, "compliance_ack_mismatch", "ack record is expired, consumed, or not bound to this article version")
|
|||
|
|
}
|
|||
|
|
return nil, response.ErrInternal(51004, "compliance_ack_lookup_failed", "failed to validate compliance ack")
|
|||
|
|
}
|
|||
|
|
record, err := s.loadCheckRecord(ctx, req.TenantID, recordID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
expectedFingerprint := policyFingerprint(policy, req.TargetPlatforms)
|
|||
|
|
if record.ArticleID != req.ArticleID ||
|
|||
|
|
record.ArticleVersionID != req.ArticleVersionID ||
|
|||
|
|
record.PolicyFingerprint != expectedFingerprint {
|
|||
|
|
return nil, response.ErrBadRequest(41003, "compliance_ack_mismatch", "ack record no longer matches current content or policy")
|
|||
|
|
}
|
|||
|
|
violations, err := s.loadViolations(ctx, req.TenantID, record.ID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
for _, violation := range violations {
|
|||
|
|
if record.EnforcementMode == "mandatory" && violation.Level == "block" {
|
|||
|
|
return nil, response.ErrConflict(41001, "compliance_blocked", "blocked violations cannot be acknowledged")
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
result := checkResultFromRecord(record, violations, sharedcompliance.GateDecisionPass)
|
|||
|
|
result.Decision = sharedcompliance.GateDecisionPass
|
|||
|
|
result.Passed = true
|
|||
|
|
return result, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) LatestCheck(ctx context.Context, tenantID, articleID int64) (*sharedcompliance.CheckResult, error) {
|
|||
|
|
var recordID int64
|
|||
|
|
err := s.pool.QueryRow(ctx, `
|
|||
|
|
SELECT id
|
|||
|
|
FROM compliance_check_records
|
|||
|
|
WHERE tenant_id = $1 AND article_id = $2
|
|||
|
|
ORDER BY checked_at DESC, id DESC
|
|||
|
|
LIMIT 1
|
|||
|
|
`, tenantID, articleID).Scan(&recordID)
|
|||
|
|
if err != nil {
|
|||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|||
|
|
return nil, response.ErrNotFound(40461, "compliance_check_not_found", "no compliance check record found")
|
|||
|
|
}
|
|||
|
|
return nil, response.ErrInternal(51005, "compliance_check_lookup_failed", "failed to load latest compliance check")
|
|||
|
|
}
|
|||
|
|
record, err := s.loadCheckRecord(ctx, tenantID, recordID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
violations, err := s.loadViolations(ctx, tenantID, recordID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
return checkResultFromRecord(record, violations, decide(record.EnforcementMode, record.HighestLevel, record.HitCount)), nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) MarkPublishJobBlockedByCompliance(ctx context.Context, jobID uuid.UUID, result *sharedcompliance.CheckResult) error {
|
|||
|
|
if s == nil || s.pool == nil || jobID == uuid.Nil {
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
var recordID any
|
|||
|
|
var reason any
|
|||
|
|
if result != nil && result.RecordID > 0 {
|
|||
|
|
recordID = result.RecordID
|
|||
|
|
reasonBytes, _ := json.Marshal(result)
|
|||
|
|
reason = string(reasonBytes)
|
|||
|
|
}
|
|||
|
|
tx, err := s.pool.Begin(ctx)
|
|||
|
|
if err != nil {
|
|||
|
|
return response.ErrInternal(51012, "compliance_publish_job_block_failed", "failed to begin compliance block transaction")
|
|||
|
|
}
|
|||
|
|
defer rollbackTx(tx)
|
|||
|
|
if _, err := tx.Exec(ctx, `
|
|||
|
|
UPDATE desktop_publish_jobs
|
|||
|
|
SET status = 'blocked_by_compliance',
|
|||
|
|
compliance_blocked_record_id = $2,
|
|||
|
|
compliance_blocked_at = NOW(),
|
|||
|
|
compliance_blocked_reason = $3,
|
|||
|
|
updated_at = NOW()
|
|||
|
|
WHERE desktop_id = $1
|
|||
|
|
AND status <> 'blocked_by_compliance'
|
|||
|
|
`, jobID, recordID, reason); err != nil {
|
|||
|
|
return response.ErrInternal(51012, "compliance_publish_job_block_failed", "failed to mark publish job blocked")
|
|||
|
|
}
|
|||
|
|
if _, err := tx.Exec(ctx, `
|
|||
|
|
UPDATE desktop_tasks
|
|||
|
|
SET status = 'aborted',
|
|||
|
|
error = jsonb_build_object(
|
|||
|
|
'message', 'compliance_blocked',
|
|||
|
|
'record_id', $2::bigint,
|
|||
|
|
'detail', $3::text
|
|||
|
|
),
|
|||
|
|
active_attempt_id = NULL,
|
|||
|
|
lease_token_hash = NULL,
|
|||
|
|
lease_expires_at = NULL,
|
|||
|
|
updated_at = NOW()
|
|||
|
|
WHERE job_id = $1
|
|||
|
|
AND kind = 'publish'
|
|||
|
|
AND status = 'queued'
|
|||
|
|
`, jobID, recordID, reason); err != nil {
|
|||
|
|
return response.ErrInternal(51012, "compliance_publish_job_block_failed", "failed to abort blocked publish tasks")
|
|||
|
|
}
|
|||
|
|
if err := tx.Commit(ctx); err != nil {
|
|||
|
|
return response.ErrInternal(51012, "compliance_publish_job_block_failed", "failed to commit compliance block")
|
|||
|
|
}
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) SubmitManualReview(ctx context.Context, req CreateManualReviewRequest) (*sharedcompliance.ManualReview, error) {
|
|||
|
|
note := strings.TrimSpace(req.Note)
|
|||
|
|
if note == "" {
|
|||
|
|
return nil, response.ErrBadRequest(40001, "invalid_params", "note is required")
|
|||
|
|
}
|
|||
|
|
if utf8.RuneCountInString(note) > 500 {
|
|||
|
|
return nil, response.ErrBadRequest(40001, "invalid_params", "note must be 500 characters or less")
|
|||
|
|
}
|
|||
|
|
record, err := s.loadCheckRecord(ctx, req.TenantID, req.CheckRecordID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
if record.ArticleID != req.ArticleID || record.ArticleVersionID <= 0 {
|
|||
|
|
return nil, response.ErrBadRequest(41009, "compliance_manual_review_invalid_check_record", "check record does not belong to this article version")
|
|||
|
|
}
|
|||
|
|
violations, err := s.loadViolations(ctx, req.TenantID, record.ID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
hasBlock := false
|
|||
|
|
for _, violation := range violations {
|
|||
|
|
if violation.Level == "block" {
|
|||
|
|
hasBlock = true
|
|||
|
|
break
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
if !hasBlock {
|
|||
|
|
return nil, response.ErrBadRequest(41009, "compliance_manual_review_invalid_check_record", "manual review requires at least one block violation")
|
|||
|
|
}
|
|||
|
|
summaryJSON, err := json.Marshal(violations)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, response.ErrInternal(51006, "manual_review_create_failed", "failed to encode manual review summary")
|
|||
|
|
}
|
|||
|
|
tx, err := s.pool.Begin(ctx)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, response.ErrInternal(51006, "manual_review_create_failed", "failed to begin manual review transaction")
|
|||
|
|
}
|
|||
|
|
defer rollbackTx(tx)
|
|||
|
|
var review sharedcompliance.ManualReview
|
|||
|
|
err = tx.QueryRow(ctx, `
|
|||
|
|
INSERT INTO compliance_manual_reviews (
|
|||
|
|
tenant_id,
|
|||
|
|
article_id,
|
|||
|
|
article_version_id,
|
|||
|
|
original_check_record_id,
|
|||
|
|
original_content_hash,
|
|||
|
|
original_policy_fingerprint,
|
|||
|
|
original_target_platforms,
|
|||
|
|
original_violation_summary,
|
|||
|
|
requested_by,
|
|||
|
|
request_note,
|
|||
|
|
status
|
|||
|
|
)
|
|||
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,'pending')
|
|||
|
|
RETURNING id, tenant_id, article_id, article_version_id, original_check_record_id, requested_by, requested_at, request_note, status
|
|||
|
|
`,
|
|||
|
|
req.TenantID,
|
|||
|
|
req.ArticleID,
|
|||
|
|
record.ArticleVersionID,
|
|||
|
|
record.ID,
|
|||
|
|
record.ContentHash,
|
|||
|
|
record.PolicyFingerprint,
|
|||
|
|
record.TargetPlatforms,
|
|||
|
|
summaryJSON,
|
|||
|
|
req.ActorID,
|
|||
|
|
note,
|
|||
|
|
).Scan(&review.ID, &review.TenantID, &review.ArticleID, &review.ArticleVersionID, &review.CheckRecordID, &review.RequestedBy, &review.RequestedAt, &review.RequestNote, &review.Status)
|
|||
|
|
if err != nil {
|
|||
|
|
if isUniqueViolation(err) {
|
|||
|
|
_ = tx.Rollback(ctx)
|
|||
|
|
status, lookupErr := s.lookupManualReviewStatus(ctx, req.TenantID, record.ArticleVersionID)
|
|||
|
|
if lookupErr == nil && status == sharedcompliance.ManualReviewStatusPending {
|
|||
|
|
return nil, response.ErrConflict(41007, "compliance_manual_review_pending", "manual review is already pending")
|
|||
|
|
}
|
|||
|
|
return nil, response.ErrConflict(41008, "compliance_manual_review_terminal", "manual review for this version is already terminal")
|
|||
|
|
}
|
|||
|
|
return nil, response.ErrInternal(51006, "manual_review_create_failed", "failed to create manual review")
|
|||
|
|
}
|
|||
|
|
if _, err := tx.Exec(ctx, `
|
|||
|
|
UPDATE article_versions
|
|||
|
|
SET manual_review_status = 'pending',
|
|||
|
|
manual_review_id = $1,
|
|||
|
|
manual_review_updated_at = NOW()
|
|||
|
|
WHERE id = $2 AND article_id = $3
|
|||
|
|
`, review.ID, record.ArticleVersionID, req.ArticleID); err != nil {
|
|||
|
|
return nil, response.ErrInternal(51006, "manual_review_create_failed", "failed to stamp article version")
|
|||
|
|
}
|
|||
|
|
if err := tx.Commit(ctx); err != nil {
|
|||
|
|
return nil, response.ErrInternal(51006, "manual_review_create_failed", "failed to commit manual review")
|
|||
|
|
}
|
|||
|
|
review.ViolationSummary = violations
|
|||
|
|
review.OriginalPlatforms = record.TargetPlatforms
|
|||
|
|
s.reviewCache.Delete(record.ArticleVersionID)
|
|||
|
|
return &review, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) LatestManualReview(ctx context.Context, tenantID, articleID int64, requestedVersionID *int64) (*sharedcompliance.ManualReview, error) {
|
|||
|
|
versionID, err := s.ResolvePublishableVersion(ctx, tenantID, articleID, requestedVersionID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
review, err := s.loadLatestManualReviewByVersion(ctx, tenantID, versionID)
|
|||
|
|
if err != nil {
|
|||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|||
|
|
return &sharedcompliance.ManualReview{
|
|||
|
|
TenantID: tenantID,
|
|||
|
|
ArticleID: articleID,
|
|||
|
|
ArticleVersionID: versionID,
|
|||
|
|
Status: sharedcompliance.ManualReviewStatusNone,
|
|||
|
|
}, nil
|
|||
|
|
}
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
return review, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) CancelManualReview(ctx context.Context, req CancelManualReviewRequest) (*sharedcompliance.ManualReview, error) {
|
|||
|
|
reason := strings.TrimSpace(req.Reason)
|
|||
|
|
if utf8.RuneCountInString(reason) > 500 {
|
|||
|
|
return nil, response.ErrBadRequest(40001, "invalid_params", "reason must be 500 characters or less")
|
|||
|
|
}
|
|||
|
|
tx, err := s.pool.Begin(ctx)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, response.ErrInternal(51007, "manual_review_cancel_failed", "failed to begin manual review cancellation")
|
|||
|
|
}
|
|||
|
|
defer rollbackTx(tx)
|
|||
|
|
var review sharedcompliance.ManualReview
|
|||
|
|
err = tx.QueryRow(ctx, `
|
|||
|
|
UPDATE compliance_manual_reviews
|
|||
|
|
SET status = 'cancelled',
|
|||
|
|
cancelled_by = $1,
|
|||
|
|
cancelled_at = NOW(),
|
|||
|
|
cancel_reason = $2
|
|||
|
|
WHERE id = $3
|
|||
|
|
AND tenant_id = $4
|
|||
|
|
AND article_id = $5
|
|||
|
|
AND requested_by = $1
|
|||
|
|
AND status = 'pending'
|
|||
|
|
RETURNING id, tenant_id, article_id, article_version_id, original_check_record_id, requested_by, requested_at, request_note, status, cancelled_by, cancelled_at, cancel_reason
|
|||
|
|
`, req.ActorID, nullableString(reason), req.ReviewID, req.TenantID, req.ArticleID).Scan(
|
|||
|
|
&review.ID,
|
|||
|
|
&review.TenantID,
|
|||
|
|
&review.ArticleID,
|
|||
|
|
&review.ArticleVersionID,
|
|||
|
|
&review.CheckRecordID,
|
|||
|
|
&review.RequestedBy,
|
|||
|
|
&review.RequestedAt,
|
|||
|
|
&review.RequestNote,
|
|||
|
|
&review.Status,
|
|||
|
|
&review.CancelledBy,
|
|||
|
|
&review.CancelledAt,
|
|||
|
|
&review.CancelReason,
|
|||
|
|
)
|
|||
|
|
if err != nil {
|
|||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|||
|
|
return nil, response.ErrConflict(41010, "compliance_manual_review_not_cancellable", "manual review cannot be cancelled")
|
|||
|
|
}
|
|||
|
|
return nil, response.ErrInternal(51007, "manual_review_cancel_failed", "failed to cancel manual review")
|
|||
|
|
}
|
|||
|
|
if _, err := tx.Exec(ctx, `
|
|||
|
|
UPDATE article_versions
|
|||
|
|
SET manual_review_status = 'none',
|
|||
|
|
manual_review_id = NULL,
|
|||
|
|
manual_review_updated_at = NOW()
|
|||
|
|
WHERE id = $1 AND article_id = $2
|
|||
|
|
`, review.ArticleVersionID, req.ArticleID); err != nil {
|
|||
|
|
return nil, response.ErrInternal(51007, "manual_review_cancel_failed", "failed to clear article version stamp")
|
|||
|
|
}
|
|||
|
|
if err := tx.Commit(ctx); err != nil {
|
|||
|
|
return nil, response.ErrInternal(51007, "manual_review_cancel_failed", "failed to commit manual review cancellation")
|
|||
|
|
}
|
|||
|
|
s.reviewCache.Delete(review.ArticleVersionID)
|
|||
|
|
return &review, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func DecodeReviewJobMessage(body []byte) (ReviewJobMessage, error) {
|
|||
|
|
var msg ReviewJobMessage
|
|||
|
|
if err := json.Unmarshal(body, &msg); err != nil {
|
|||
|
|
return msg, err
|
|||
|
|
}
|
|||
|
|
if msg.JobID <= 0 || msg.TenantID <= 0 || msg.ArticleID <= 0 || msg.ArticleVersionID <= 0 || msg.CheckRecordID <= 0 {
|
|||
|
|
return msg, fmt.Errorf("invalid compliance review job message")
|
|||
|
|
}
|
|||
|
|
return msg, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) ProcessReviewJob(ctx context.Context, msg ReviewJobMessage) error {
|
|||
|
|
if s == nil || s.pool == nil {
|
|||
|
|
return response.ErrServiceUnavailable(41004, "compliance_engine_unavailable", "compliance service is unavailable")
|
|||
|
|
}
|
|||
|
|
cfg := s.currentComplianceConfig()
|
|||
|
|
if !cfg.Enabled || !cfg.LLMJudgeEnabled {
|
|||
|
|
return s.markReviewJobDone(ctx, msg.JobID)
|
|||
|
|
}
|
|||
|
|
if s.llm == nil || s.llm.Validate() != nil {
|
|||
|
|
return s.failReviewJob(ctx, msg.JobID, cfg.ReviewMaxAttempts, sharedllm.ErrNotConfigured)
|
|||
|
|
}
|
|||
|
|
job, claimed, err := s.claimReviewJob(ctx, msg)
|
|||
|
|
if err != nil || !claimed {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
prompt, err := s.buildReviewPrompt(ctx, job)
|
|||
|
|
if err != nil {
|
|||
|
|
return s.failReviewJob(ctx, job.ID, cfg.ReviewMaxAttempts, err)
|
|||
|
|
}
|
|||
|
|
if strings.TrimSpace(prompt) == "" {
|
|||
|
|
return s.markReviewJobDone(ctx, job.ID)
|
|||
|
|
}
|
|||
|
|
reservation, billingErr := s.reserveComplianceLLMJudgePoints(ctx, job, prompt)
|
|||
|
|
if billingErr != nil {
|
|||
|
|
if s.logger != nil {
|
|||
|
|
s.logger.Warn("compliance llm billing skipped review",
|
|||
|
|
zap.Error(billingErr),
|
|||
|
|
zap.Int64("tenant_id", job.TenantID),
|
|||
|
|
zap.Int64("job_id", job.ID),
|
|||
|
|
zap.Int64("check_record_id", job.CheckRecordID),
|
|||
|
|
)
|
|||
|
|
}
|
|||
|
|
return s.markReviewJobDone(ctx, job.ID)
|
|||
|
|
}
|
|||
|
|
result, err := s.llm.Generate(ctx, sharedllm.GenerateRequest{
|
|||
|
|
Prompt: prompt,
|
|||
|
|
Timeout: cfg.ReviewWorkerTimeout,
|
|||
|
|
MaxOutputTokens: 900,
|
|||
|
|
ResponseFormat: &sharedllm.ResponseFormat{
|
|||
|
|
Type: sharedllm.ResponseFormatTypeJSONObject,
|
|||
|
|
Name: "compliance_review",
|
|||
|
|
Description: "Semantic compliance review result for a publish check record.",
|
|||
|
|
},
|
|||
|
|
}, nil)
|
|||
|
|
if err != nil {
|
|||
|
|
_ = s.refundComplianceLLMJudgePoints(context.Background(), job, reservation, err)
|
|||
|
|
return s.failReviewJob(ctx, job.ID, cfg.ReviewMaxAttempts, err)
|
|||
|
|
}
|
|||
|
|
summary := decodeLLMReviewResult(result.Content)
|
|||
|
|
if err := s.persistLLMReviewResult(ctx, job, summary); err != nil {
|
|||
|
|
_ = s.refundComplianceLLMJudgePoints(context.Background(), job, reservation, err)
|
|||
|
|
return s.failReviewJob(ctx, job.ID, cfg.ReviewMaxAttempts, err)
|
|||
|
|
}
|
|||
|
|
if err := s.completeComplianceLLMJudgePoints(ctx, job, reservation, result.Model); err != nil && s.logger != nil {
|
|||
|
|
s.logger.Warn("compliance llm billing completion failed",
|
|||
|
|
zap.Error(err),
|
|||
|
|
zap.Int64("tenant_id", job.TenantID),
|
|||
|
|
zap.Int64("job_id", job.ID),
|
|||
|
|
zap.Int64("check_record_id", job.CheckRecordID),
|
|||
|
|
)
|
|||
|
|
}
|
|||
|
|
return s.markReviewJobDone(ctx, job.ID)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type reviewJob struct {
|
|||
|
|
ID int64
|
|||
|
|
TenantID int64
|
|||
|
|
ArticleID int64
|
|||
|
|
ArticleVersionID int64
|
|||
|
|
CheckRecordID int64
|
|||
|
|
CheckedBy int64
|
|||
|
|
Attempts int
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) claimReviewJob(ctx context.Context, msg ReviewJobMessage) (reviewJob, bool, error) {
|
|||
|
|
var job reviewJob
|
|||
|
|
err := s.pool.QueryRow(ctx, `
|
|||
|
|
UPDATE compliance_review_jobs
|
|||
|
|
SET status = 'running',
|
|||
|
|
attempts = attempts + 1,
|
|||
|
|
locked_at = NOW(),
|
|||
|
|
locked_by = $2,
|
|||
|
|
updated_at = NOW()
|
|||
|
|
WHERE id = $1
|
|||
|
|
AND status IN ('pending','failed')
|
|||
|
|
AND available_at <= NOW()
|
|||
|
|
RETURNING
|
|||
|
|
id,
|
|||
|
|
tenant_id,
|
|||
|
|
article_id,
|
|||
|
|
article_version_id,
|
|||
|
|
check_record_id,
|
|||
|
|
COALESCE((
|
|||
|
|
SELECT checked_by
|
|||
|
|
FROM compliance_check_records cr
|
|||
|
|
WHERE cr.id = compliance_review_jobs.check_record_id
|
|||
|
|
), 0),
|
|||
|
|
attempts
|
|||
|
|
`, msg.JobID, strings.TrimSpace(msg.MessageID)).Scan(
|
|||
|
|
&job.ID,
|
|||
|
|
&job.TenantID,
|
|||
|
|
&job.ArticleID,
|
|||
|
|
&job.ArticleVersionID,
|
|||
|
|
&job.CheckRecordID,
|
|||
|
|
&job.CheckedBy,
|
|||
|
|
&job.Attempts,
|
|||
|
|
)
|
|||
|
|
if err != nil {
|
|||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|||
|
|
return job, false, nil
|
|||
|
|
}
|
|||
|
|
return job, false, err
|
|||
|
|
}
|
|||
|
|
return job, true, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) buildReviewPrompt(ctx context.Context, job reviewJob) (string, error) {
|
|||
|
|
var title string
|
|||
|
|
var plaintext pgtype.Text
|
|||
|
|
err := s.pool.QueryRow(ctx, `
|
|||
|
|
SELECT av.title, av.plaintext_snapshot
|
|||
|
|
FROM article_versions av
|
|||
|
|
JOIN articles a ON a.id = av.article_id
|
|||
|
|
WHERE av.id = $1 AND a.id = $2 AND a.tenant_id = $3
|
|||
|
|
`, job.ArticleVersionID, job.ArticleID, job.TenantID).Scan(&title, &plaintext)
|
|||
|
|
if err != nil {
|
|||
|
|
return "", err
|
|||
|
|
}
|
|||
|
|
text := strings.TrimSpace(plaintext.String)
|
|||
|
|
if text == "" {
|
|||
|
|
snapshot, err := s.loadArticleVersionSnapshot(ctx, job.TenantID, job.ArticleID, job.ArticleVersionID)
|
|||
|
|
if err != nil {
|
|||
|
|
return "", err
|
|||
|
|
}
|
|||
|
|
text = snapshot.Plaintext
|
|||
|
|
title = snapshot.Title
|
|||
|
|
}
|
|||
|
|
return strings.TrimSpace(`你是内容合规复核助手。请只基于给定标题和正文判断是否存在广告法极限词、政治敏感、涉黄涉暴、行业违禁或明显平台高风险表达。
|
|||
|
|
|
|||
|
|
返回 JSON，不要输出 Markdown：
|
|||
|
|
{
|
|||
|
|
"risk": true,
|
|||
|
|
"level": "high",
|
|||
|
|
"summary": "一句话说明风险",
|
|||
|
|
"evidence": "命中的短句或关键词",
|
|||
|
|
"hint": "给作者的修改建议"
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
若没有高风险，返回 {"risk": false, "level": "info", "summary": "", "evidence": "", "hint": ""}。
|
|||
|
|
|
|||
|
|
标题：` + title + `
|
|||
|
|
正文：
|
|||
|
|
` + truncateRunes(text, 6000)), nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type llmReviewSummary struct {
|
|||
|
|
Risk bool `json:"risk"`
|
|||
|
|
Level string `json:"level"`
|
|||
|
|
Summary string `json:"summary"`
|
|||
|
|
Evidence string `json:"evidence"`
|
|||
|
|
Hint string `json:"hint"`
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func decodeLLMReviewResult(content string) llmReviewSummary {
|
|||
|
|
var summary llmReviewSummary
|
|||
|
|
_ = json.Unmarshal([]byte(strings.TrimSpace(content)), &summary)
|
|||
|
|
summary.Level = normalizeLevel(summary.Level)
|
|||
|
|
if summary.Level == "" {
|
|||
|
|
summary.Level = "info"
|
|||
|
|
}
|
|||
|
|
summary.Summary = strings.TrimSpace(summary.Summary)
|
|||
|
|
summary.Evidence = strings.TrimSpace(summary.Evidence)
|
|||
|
|
summary.Hint = strings.TrimSpace(summary.Hint)
|
|||
|
|
return summary
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) persistLLMReviewResult(ctx context.Context, job reviewJob, summary llmReviewSummary) error {
|
|||
|
|
if !summary.Risk {
|
|||
|
|
_, err := s.pool.Exec(ctx, `
|
|||
|
|
UPDATE compliance_check_records
|
|||
|
|
SET llm_used = TRUE
|
|||
|
|
WHERE id = $1 AND tenant_id = $2
|
|||
|
|
`, job.CheckRecordID, job.TenantID)
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
matchedText := summary.Evidence
|
|||
|
|
if matchedText == "" {
|
|||
|
|
matchedText = summary.Summary
|
|||
|
|
}
|
|||
|
|
hint := summary.Hint
|
|||
|
|
if hint == "" {
|
|||
|
|
hint = summary.Summary
|
|||
|
|
}
|
|||
|
|
tx, err := s.pool.Begin(ctx)
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
defer rollbackTx(tx)
|
|||
|
|
if _, err := tx.Exec(ctx, `
|
|||
|
|
INSERT INTO compliance_violations (
|
|||
|
|
record_id,
|
|||
|
|
tenant_id,
|
|||
|
|
source,
|
|||
|
|
matched_text,
|
|||
|
|
start_offset,
|
|||
|
|
end_offset,
|
|||
|
|
level,
|
|||
|
|
hint
|
|||
|
|
)
|
|||
|
|
VALUES ($1,$2,'llm',$3,0,0,$4,$5)
|
|||
|
|
`, job.CheckRecordID, job.TenantID, truncateRunes(matchedText, maxMatchedTextRunes), summary.Level, nullableString(hint)); err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
if _, err := tx.Exec(ctx, `
|
|||
|
|
UPDATE compliance_check_records
|
|||
|
|
SET llm_used = TRUE,
|
|||
|
|
hit_count = hit_count + 1,
|
|||
|
|
stored_hit_count = stored_hit_count + 1,
|
|||
|
|
highest_level = CASE
|
|||
|
|
WHEN highest_level IS NULL OR $3 > CASE highest_level
|
|||
|
|
WHEN 'block' THEN 4
|
|||
|
|
WHEN 'high' THEN 3
|
|||
|
|
WHEN 'medium' THEN 2
|
|||
|
|
WHEN 'info' THEN 1
|
|||
|
|
ELSE 0
|
|||
|
|
END THEN $2
|
|||
|
|
ELSE highest_level
|
|||
|
|
END
|
|||
|
|
WHERE id = $1 AND tenant_id = $4
|
|||
|
|
`, job.CheckRecordID, summary.Level, levelRank(summary.Level), job.TenantID); err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
return tx.Commit(ctx)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
type aiPointReservation struct {
|
|||
|
|
ReservationID int64
|
|||
|
|
UsageLogID int64
|
|||
|
|
Points int
|
|||
|
|
RequestChars int
|
|||
|
|
BaseChars int
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) reserveComplianceLLMJudgePoints(ctx context.Context, job reviewJob, meteredText string) (*aiPointReservation, error) {
|
|||
|
|
if s == nil || s.pool == nil {
|
|||
|
|
return nil, response.ErrServiceUnavailable(50344, "ai_points_unavailable", "ai points store is unavailable")
|
|||
|
|
}
|
|||
|
|
tx, err := s.pool.Begin(ctx)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
defer rollbackTx(tx)
|
|||
|
|
if err := lockTenantAIPoints(ctx, tx, job.TenantID); err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
now := time.Now().UTC()
|
|||
|
|
quotaRepo := tenantrepo.NewQuotaRepository(tx)
|
|||
|
|
status, err := quotaRepo.GetAIQuotaStatus(ctx, job.TenantID, now)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
requestChars := countAIPointChars(meteredText)
|
|||
|
|
points := calculateAIPointCost(requestChars, status.BaseChars)
|
|||
|
|
if status.Total <= 0 || status.Balance < points {
|
|||
|
|
return nil, response.ErrForbidden(40381, "ai_points_insufficient", "AI points are insufficient")
|
|||
|
|
}
|
|||
|
|
resourceType := aiResourceTypeComplianceReview
|
|||
|
|
resourceID := job.CheckRecordID
|
|||
|
|
reservationID, err := quotaRepo.CreateQuotaReservation(ctx, tenantrepo.QuotaReservationInput{
|
|||
|
|
TenantID: job.TenantID,
|
|||
|
|
OperatorID: job.CheckedBy,
|
|||
|
|
QuotaType: aiPointsQuotaType,
|
|||
|
|
ResourceType: resourceType,
|
|||
|
|
ResourceID: &resourceID,
|
|||
|
|
ReservedAmount: points,
|
|||
|
|
ExpireAt: now.Add(time.Hour),
|
|||
|
|
})
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
metadataJSON, err := json.Marshal(map[string]any{
|
|||
|
|
"article_id": job.ArticleID,
|
|||
|
|
"article_version_id": job.ArticleVersionID,
|
|||
|
|
"check_record_id": job.CheckRecordID,
|
|||
|
|
"review_job_id": job.ID,
|
|||
|
|
})
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
usageID, err := tenantrepo.NewAIPointUsageRepository(tx).Create(ctx, tenantrepo.CreateAIPointUsageLogInput{
|
|||
|
|
TenantID: job.TenantID,
|
|||
|
|
OperatorID: job.CheckedBy,
|
|||
|
|
QuotaReservationID: reservationID,
|
|||
|
|
UsageType: aiUsageTypeComplianceLLMJudge,
|
|||
|
|
ResourceType: &resourceType,
|
|||
|
|
ResourceID: &resourceID,
|
|||
|
|
RequestChars: requestChars,
|
|||
|
|
BaseChars: status.BaseChars,
|
|||
|
|
Points: points,
|
|||
|
|
MetadataJSON: metadataJSON,
|
|||
|
|
})
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
reason := "ai_points_reserve"
|
|||
|
|
referenceType := "ai_point_usage"
|
|||
|
|
if _, err := quotaRepo.InsertQuotaLedger(ctx, tenantrepo.QuotaLedgerInput{
|
|||
|
|
TenantID: job.TenantID,
|
|||
|
|
OperatorID: job.CheckedBy,
|
|||
|
|
QuotaType: aiPointsQuotaType,
|
|||
|
|
Delta: -points,
|
|||
|
|
BalanceAfter: status.Balance - points,
|
|||
|
|
Reason: &reason,
|
|||
|
|
ReferenceType: &referenceType,
|
|||
|
|
ReferenceID: &usageID,
|
|||
|
|
}); err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
if err := tx.Commit(ctx); err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
return &aiPointReservation{
|
|||
|
|
ReservationID: reservationID,
|
|||
|
|
UsageLogID: usageID,
|
|||
|
|
Points: points,
|
|||
|
|
RequestChars: requestChars,
|
|||
|
|
BaseChars: status.BaseChars,
|
|||
|
|
}, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) completeComplianceLLMJudgePoints(ctx context.Context, job reviewJob, reservation *aiPointReservation, model string) error {
|
|||
|
|
if s == nil || s.pool == nil || reservation == nil || reservation.ReservationID <= 0 || reservation.UsageLogID <= 0 {
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
tx, err := s.pool.Begin(ctx)
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
defer rollbackTx(tx)
|
|||
|
|
pending, err := lockPendingAIPointUsage(ctx, tx, job.TenantID, reservation.UsageLogID)
|
|||
|
|
if err != nil || !pending {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
if err := confirmAIPointReservation(ctx, tx, job.TenantID, reservation.ReservationID); err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
var modelPtr *string
|
|||
|
|
if strings.TrimSpace(model) != "" {
|
|||
|
|
normalized := strings.TrimSpace(model)
|
|||
|
|
modelPtr = &normalized
|
|||
|
|
}
|
|||
|
|
if err := tenantrepo.NewAIPointUsageRepository(tx).MarkCompleted(ctx, job.TenantID, reservation.UsageLogID, modelPtr); err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
return tx.Commit(ctx)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) refundComplianceLLMJudgePoints(ctx context.Context, job reviewJob, reservation *aiPointReservation, jobErr error) error {
|
|||
|
|
if s == nil || s.pool == nil || reservation == nil || reservation.ReservationID <= 0 || reservation.UsageLogID <= 0 || reservation.Points <= 0 {
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
tx, err := s.pool.Begin(ctx)
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
defer rollbackTx(tx)
|
|||
|
|
if err := lockTenantAIPoints(ctx, tx, job.TenantID); err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
pending, err := lockPendingAIPointUsage(ctx, tx, job.TenantID, reservation.UsageLogID)
|
|||
|
|
if err != nil || !pending {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
status, err := tenantrepo.NewQuotaRepository(tx).GetAIQuotaStatus(ctx, job.TenantID, time.Now().UTC())
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
quotaRepo := tenantrepo.NewQuotaRepository(tx)
|
|||
|
|
reason := "ai_points_refund"
|
|||
|
|
referenceType := "ai_point_usage"
|
|||
|
|
usageID := reservation.UsageLogID
|
|||
|
|
if _, err := quotaRepo.InsertQuotaLedger(ctx, tenantrepo.QuotaLedgerInput{
|
|||
|
|
TenantID: job.TenantID,
|
|||
|
|
OperatorID: job.CheckedBy,
|
|||
|
|
QuotaType: aiPointsQuotaType,
|
|||
|
|
Delta: reservation.Points,
|
|||
|
|
BalanceAfter: status.Balance + reservation.Points,
|
|||
|
|
Reason: &reason,
|
|||
|
|
ReferenceType: &referenceType,
|
|||
|
|
ReferenceID: &usageID,
|
|||
|
|
}); err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
if err := refundAIPointReservation(ctx, tx, job.TenantID, reservation.ReservationID); err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
if err := tenantrepo.NewAIPointUsageRepository(tx).MarkRefunded(ctx, job.TenantID, reservation.UsageLogID, truncateRunes(errorString(jobErr), 2000)); err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
return tx.Commit(ctx)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func countAIPointChars(text string) int {
|
|||
|
|
return utf8.RuneCountInString(strings.TrimSpace(text))
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func calculateAIPointCost(chars, baseChars int) int {
|
|||
|
|
if baseChars <= 0 {
|
|||
|
|
baseChars = 1000
|
|||
|
|
}
|
|||
|
|
if chars <= 0 {
|
|||
|
|
return 1
|
|||
|
|
}
|
|||
|
|
return chars/baseChars + 1
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func lockTenantAIPoints(ctx context.Context, tx pgx.Tx, tenantID int64) error {
|
|||
|
|
_, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(20260428, hashint8($1)::int)`, tenantID)
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func lockPendingAIPointUsage(ctx context.Context, tx pgx.Tx, tenantID, usageLogID int64) (bool, error) {
|
|||
|
|
var status string
|
|||
|
|
err := tx.QueryRow(ctx, `
|
|||
|
|
SELECT status
|
|||
|
|
FROM ai_point_usage_logs
|
|||
|
|
WHERE id = $1 AND tenant_id = $2
|
|||
|
|
FOR UPDATE
|
|||
|
|
`, usageLogID, tenantID).Scan(&status)
|
|||
|
|
if err != nil {
|
|||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|||
|
|
return false, nil
|
|||
|
|
}
|
|||
|
|
return false, err
|
|||
|
|
}
|
|||
|
|
return status == "pending", nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func confirmAIPointReservation(ctx context.Context, tx pgx.Tx, tenantID, reservationID int64) error {
|
|||
|
|
tag, err := tx.Exec(ctx, `
|
|||
|
|
UPDATE quota_reservations
|
|||
|
|
SET status = 'confirmed',
|
|||
|
|
consumed_amount = reserved_amount,
|
|||
|
|
updated_at = NOW()
|
|||
|
|
WHERE id = $1
|
|||
|
|
AND tenant_id = $2
|
|||
|
|
AND quota_type = $3
|
|||
|
|
AND status = 'pending'
|
|||
|
|
`, reservationID, tenantID, aiPointsQuotaType)
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
if tag.RowsAffected() == 0 {
|
|||
|
|
return errors.New("ai points reservation is not pending")
|
|||
|
|
}
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func refundAIPointReservation(ctx context.Context, tx pgx.Tx, tenantID, reservationID int64) error {
|
|||
|
|
tag, err := tx.Exec(ctx, `
|
|||
|
|
UPDATE quota_reservations
|
|||
|
|
SET status = 'refunded',
|
|||
|
|
refunded_amount = reserved_amount,
|
|||
|
|
updated_at = NOW()
|
|||
|
|
WHERE id = $1
|
|||
|
|
AND tenant_id = $2
|
|||
|
|
AND quota_type = $3
|
|||
|
|
AND status = 'pending'
|
|||
|
|
`, reservationID, tenantID, aiPointsQuotaType)
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
if tag.RowsAffected() == 0 {
|
|||
|
|
return errors.New("ai points reservation is not pending")
|
|||
|
|
}
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) markReviewJobDone(ctx context.Context, jobID int64) error {
|
|||
|
|
_, err := s.pool.Exec(ctx, `
|
|||
|
|
UPDATE compliance_review_jobs
|
|||
|
|
SET status = 'done',
|
|||
|
|
locked_at = NULL,
|
|||
|
|
locked_by = NULL,
|
|||
|
|
updated_at = NOW()
|
|||
|
|
WHERE id = $1
|
|||
|
|
`, jobID)
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) failReviewJob(ctx context.Context, jobID int64, maxAttempts int, jobErr error) error {
|
|||
|
|
if maxAttempts <= 0 {
|
|||
|
|
maxAttempts = 3
|
|||
|
|
}
|
|||
|
|
_, err := s.pool.Exec(ctx, `
|
|||
|
|
UPDATE compliance_review_jobs
|
|||
|
|
SET status = CASE WHEN attempts >= $2 THEN 'failed' ELSE 'pending' END,
|
|||
|
|
available_at = CASE WHEN attempts >= $2 THEN available_at ELSE NOW() + interval '5 minutes' END,
|
|||
|
|
locked_at = NULL,
|
|||
|
|
locked_by = NULL,
|
|||
|
|
last_error = $3,
|
|||
|
|
updated_at = NOW()
|
|||
|
|
WHERE id = $1
|
|||
|
|
`, jobID, maxAttempts, truncateRunes(errorString(jobErr), 2000))
|
|||
|
|
if err != nil {
|
|||
|
|
return err
|
|||
|
|
}
|
|||
|
|
return jobErr
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) enqueueLLMReview(ctx context.Context, input runInput, checkRecordID int64) {
|
|||
|
|
if s == nil || s.pool == nil || s.mq == nil || checkRecordID <= 0 {
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
messageID := uuid.New()
|
|||
|
|
var jobID int64
|
|||
|
|
err := s.pool.QueryRow(ctx, `
|
|||
|
|
INSERT INTO compliance_review_jobs (
|
|||
|
|
message_id,
|
|||
|
|
tenant_id,
|
|||
|
|
article_id,
|
|||
|
|
article_version_id,
|
|||
|
|
check_record_id,
|
|||
|
|
status
|
|||
|
|
)
|
|||
|
|
VALUES ($1,$2,$3,$4,$5,'pending')
|
|||
|
|
ON CONFLICT (check_record_id) DO UPDATE
|
|||
|
|
SET available_at = LEAST(compliance_review_jobs.available_at, NOW()),
|
|||
|
|
updated_at = NOW()
|
|||
|
|
RETURNING id
|
|||
|
|
`, messageID, input.TenantID, input.ArticleID, input.ArticleVersionID, checkRecordID).Scan(&jobID)
|
|||
|
|
if err != nil {
|
|||
|
|
if s.logger != nil {
|
|||
|
|
s.logger.Warn("compliance review job enqueue failed",
|
|||
|
|
zap.Error(err),
|
|||
|
|
zap.Int64("tenant_id", input.TenantID),
|
|||
|
|
zap.Int64("article_id", input.ArticleID),
|
|||
|
|
zap.Int64("check_record_id", checkRecordID),
|
|||
|
|
)
|
|||
|
|
}
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
message := ReviewJobMessage{
|
|||
|
|
JobID: jobID,
|
|||
|
|
MessageID: messageID.String(),
|
|||
|
|
TenantID: input.TenantID,
|
|||
|
|
ArticleID: input.ArticleID,
|
|||
|
|
ArticleVersionID: input.ArticleVersionID,
|
|||
|
|
CheckRecordID: checkRecordID,
|
|||
|
|
}
|
|||
|
|
body, err := json.Marshal(message)
|
|||
|
|
if err != nil {
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
if err := s.mq.PublishComplianceReviewTask(ctx, body); err != nil && s.logger != nil {
|
|||
|
|
s.logger.Warn("compliance review message publish failed",
|
|||
|
|
zap.Error(err),
|
|||
|
|
zap.Int64("job_id", jobID),
|
|||
|
|
zap.Int64("check_record_id", checkRecordID),
|
|||
|
|
)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) createManualReviewBypassRecord(ctx context.Context, req GateForPublishRequest, snapshot articleVersionSnapshot) (*sharedcompliance.CheckResult, error) {
|
|||
|
|
policy := &policySnapshot{
|
|||
|
|
MasterEnabled: true,
|
|||
|
|
EnforcementMode: "disabled",
|
|||
|
|
DictVersions: map[string]int{},
|
|||
|
|
}
|
|||
|
|
return s.runAndPersist(ctx, runInput{
|
|||
|
|
TenantID: req.TenantID,
|
|||
|
|
ArticleID: req.ArticleID,
|
|||
|
|
ArticleVersionID: req.ArticleVersionID,
|
|||
|
|
ActorID: req.ActorID,
|
|||
|
|
Title: snapshot.Title,
|
|||
|
|
Plaintext: "",
|
|||
|
|
TargetPlatforms: req.TargetPlatforms,
|
|||
|
|
TriggerSource: triggerManualReviewBypass,
|
|||
|
|
Policy: policy,
|
|||
|
|
ManualStatus: snapshot.ManualReviewStatus,
|
|||
|
|
ManualReviewID: snapshot.ManualReviewID,
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) loadArticleVersionSnapshot(ctx context.Context, tenantID, articleID, versionID int64) (articleVersionSnapshot, error) {
|
|||
|
|
var snapshot articleVersionSnapshot
|
|||
|
|
var title pgtype.Text
|
|||
|
|
var plaintext pgtype.Text
|
|||
|
|
var manualStatus string
|
|||
|
|
var manualReviewID pgtype.Int8
|
|||
|
|
var decisionReason pgtype.Text
|
|||
|
|
var reviewedAt pgtype.Timestamptz
|
|||
|
|
err := s.pool.QueryRow(ctx, `
|
|||
|
|
SELECT
|
|||
|
|
a.id,
|
|||
|
|
av.id,
|
|||
|
|
av.title,
|
|||
|
|
av.plaintext_snapshot,
|
|||
|
|
av.manual_review_status,
|
|||
|
|
av.manual_review_id,
|
|||
|
|
mr.decision_reason,
|
|||
|
|
mr.reviewed_at
|
|||
|
|
FROM article_versions av
|
|||
|
|
JOIN articles a ON a.id = av.article_id
|
|||
|
|
LEFT JOIN compliance_manual_reviews mr ON mr.id = av.manual_review_id
|
|||
|
|
WHERE a.tenant_id = $1
|
|||
|
|
AND a.id = $2
|
|||
|
|
AND av.id = $3
|
|||
|
|
AND a.deleted_at IS NULL
|
|||
|
|
`, tenantID, articleID, versionID).Scan(
|
|||
|
|
&snapshot.ArticleID,
|
|||
|
|
&snapshot.ArticleVersionID,
|
|||
|
|
&title,
|
|||
|
|
&plaintext,
|
|||
|
|
&manualStatus,
|
|||
|
|
&manualReviewID,
|
|||
|
|
&decisionReason,
|
|||
|
|
&reviewedAt,
|
|||
|
|
)
|
|||
|
|
if err != nil {
|
|||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|||
|
|
return snapshot, response.ErrBadRequest(41005, "invalid_article_version", "article version does not belong to this article")
|
|||
|
|
}
|
|||
|
|
return snapshot, err
|
|||
|
|
}
|
|||
|
|
snapshot.Title = strings.TrimSpace(title.String)
|
|||
|
|
snapshot.Plaintext = strings.TrimSpace(plaintext.String)
|
|||
|
|
if snapshot.Plaintext == "" {
|
|||
|
|
content, err := tenantrepo.LoadArticleVersionContent(ctx, s.pool, articleID, versionID)
|
|||
|
|
if err == nil && content != nil {
|
|||
|
|
snapshot.Plaintext = strings.TrimSpace(snapshot.Title + "\n" + tenantrepo.ExtractArticlePlaintext(stringValue(content.MarkdownContent)))
|
|||
|
|
_, _ = s.pool.Exec(ctx, `
|
|||
|
|
UPDATE article_versions SET plaintext_snapshot = $1 WHERE id = $2 AND plaintext_snapshot IS NULL
|
|||
|
|
`, snapshot.Plaintext, versionID)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
snapshot.ManualReviewStatus = sharedcompliance.ManualReviewStatus(strings.TrimSpace(manualStatus))
|
|||
|
|
if snapshot.ManualReviewStatus == "" {
|
|||
|
|
snapshot.ManualReviewStatus = sharedcompliance.ManualReviewStatusNone
|
|||
|
|
}
|
|||
|
|
if manualReviewID.Valid {
|
|||
|
|
value := manualReviewID.Int64
|
|||
|
|
snapshot.ManualReviewID = &value
|
|||
|
|
}
|
|||
|
|
if decisionReason.Valid {
|
|||
|
|
value := decisionReason.String
|
|||
|
|
snapshot.ManualReviewReason = &value
|
|||
|
|
}
|
|||
|
|
if reviewedAt.Valid {
|
|||
|
|
value := reviewedAt.Time
|
|||
|
|
snapshot.ManualReviewDecidedAt = &value
|
|||
|
|
}
|
|||
|
|
return snapshot, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) resolvePolicy(ctx context.Context, _ int64) (*policySnapshot, error) {
|
|||
|
|
cfg := s.currentComplianceConfig()
|
|||
|
|
now := time.Now()
|
|||
|
|
s.policyMu.Lock()
|
|||
|
|
if cached, ok := s.policies[globalPolicyCacheKey]; ok && cached.policy != nil && now.Before(cached.expires) {
|
|||
|
|
policy := cached.policy
|
|||
|
|
s.policyMu.Unlock()
|
|||
|
|
return policy, nil
|
|||
|
|
}
|
|||
|
|
s.policyMu.Unlock()
|
|||
|
|
|
|||
|
|
policy, err := s.loadPolicy(ctx)
|
|||
|
|
if err != nil {
|
|||
|
|
s.policyMu.Lock()
|
|||
|
|
cached := s.policies[globalPolicyCacheKey]
|
|||
|
|
s.policyMu.Unlock()
|
|||
|
|
if cached.policy != nil && now.Sub(cached.policy.LoadedAt) < 24*time.Hour {
|
|||
|
|
if s.logger != nil {
|
|||
|
|
s.logger.Warn("using stale compliance policy", zap.Error(err))
|
|||
|
|
}
|
|||
|
|
return cached.policy, nil
|
|||
|
|
}
|
|||
|
|
return nil, response.ErrServiceUnavailable(41004, "compliance_engine_unavailable", "compliance policy is unavailable")
|
|||
|
|
}
|
|||
|
|
s.policyMu.Lock()
|
|||
|
|
if s.policies == nil {
|
|||
|
|
s.policies = make(map[int64]cachedPolicy)
|
|||
|
|
}
|
|||
|
|
s.policies[globalPolicyCacheKey] = cachedPolicy{policy: policy, expires: now.Add(cfg.SnapshotReloadInterval)}
|
|||
|
|
s.policyMu.Unlock()
|
|||
|
|
return policy, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) loadPolicy(ctx context.Context) (*policySnapshot, error) {
|
|||
|
|
var (
|
|||
|
|
globalMaster bool
|
|||
|
|
globalMode string
|
|||
|
|
globalEnabledRawText string
|
|||
|
|
globalLLM bool
|
|||
|
|
globalRevision int64
|
|||
|
|
)
|
|||
|
|
err := s.pool.QueryRow(ctx, `
|
|||
|
|
SELECT
|
|||
|
|
master_enabled,
|
|||
|
|
enforcement_mode,
|
|||
|
|
enabled_dictionaries::text,
|
|||
|
|
llm_judge_enabled,
|
|||
|
|
revision
|
|||
|
|
FROM ops.compliance_policies
|
|||
|
|
WHERE id = 1
|
|||
|
|
LIMIT 1
|
|||
|
|
`).Scan(
|
|||
|
|
&globalMaster,
|
|||
|
|
&globalMode,
|
|||
|
|
&globalEnabledRawText,
|
|||
|
|
&globalLLM,
|
|||
|
|
&globalRevision,
|
|||
|
|
)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
mode := strings.TrimSpace(globalMode)
|
|||
|
|
enabledRaw := []byte(globalEnabledRawText)
|
|||
|
|
llmEnabled := globalLLM
|
|||
|
|
revision := globalRevision
|
|||
|
|
enabledIDs := decodeJSONInt64Slice(enabledRaw)
|
|||
|
|
terms, dictNames, dictVersions, versionSum, err := s.loadTerms(ctx, enabledIDs)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
return &policySnapshot{
|
|||
|
|
LoadedAt: time.Now(),
|
|||
|
|
MasterEnabled: globalMaster,
|
|||
|
|
EnforcementMode: normalizeMode(mode),
|
|||
|
|
LLMJudgeEnabled: llmEnabled && s.currentComplianceConfig().LLMJudgeEnabled,
|
|||
|
|
Revision: revision,
|
|||
|
|
DictionaryVersionSum: versionSum,
|
|||
|
|
EnabledDictionaries: dictNames,
|
|||
|
|
DictVersions: dictVersions,
|
|||
|
|
Terms: terms,
|
|||
|
|
}, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func editorPreviewPolicy(policy *policySnapshot) *policySnapshot {
|
|||
|
|
if policy == nil {
|
|||
|
|
return &policySnapshot{
|
|||
|
|
LoadedAt: time.Now(),
|
|||
|
|
MasterEnabled: true,
|
|||
|
|
EnforcementMode: "advisory",
|
|||
|
|
DictVersions: map[string]int{},
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
copyPolicy := *policy
|
|||
|
|
if !copyPolicy.MasterEnabled {
|
|||
|
|
copyPolicy.MasterEnabled = true
|
|||
|
|
}
|
|||
|
|
if copyPolicy.EnforcementMode == "disabled" || strings.TrimSpace(copyPolicy.EnforcementMode) == "" {
|
|||
|
|
copyPolicy.EnforcementMode = "advisory"
|
|||
|
|
}
|
|||
|
|
copyPolicy.LLMJudgeEnabled = false
|
|||
|
|
return ©Policy
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func normalizeArticleCheckTrigger(trigger string) string {
|
|||
|
|
switch strings.TrimSpace(trigger) {
|
|||
|
|
case triggerPublishGate:
|
|||
|
|
return triggerPublishGate
|
|||
|
|
default:
|
|||
|
|
return triggerEditorManual
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) loadTerms(ctx context.Context, enabledIDs []int64) ([]termRule, []string, map[string]int, int64, error) {
|
|||
|
|
filterEnabled := len(enabledIDs) > 0
|
|||
|
|
rows, err := s.pool.Query(ctx, `
|
|||
|
|
SELECT
|
|||
|
|
d.id,
|
|||
|
|
d.code,
|
|||
|
|
d.name,
|
|||
|
|
d.version,
|
|||
|
|
d.platform_scope,
|
|||
|
|
COALESCE(d.applicable_platforms, '{}') AS applicable_platforms,
|
|||
|
|
t.match_type,
|
|||
|
|
t.pattern,
|
|||
|
|
COALESCE(t.level_override, d.default_level) AS level,
|
|||
|
|
t.hint,
|
|||
|
|
t.reference_law
|
|||
|
|
FROM ops.compliance_dictionaries d
|
|||
|
|
JOIN ops.compliance_dictionary_terms t ON t.dictionary_id = d.id
|
|||
|
|
WHERE d.enabled = TRUE
|
|||
|
|
AND t.enabled = TRUE
|
|||
|
|
AND (NOT $1::boolean OR d.id = ANY($2::bigint[]))
|
|||
|
|
ORDER BY d.id, t.id
|
|||
|
|
`, filterEnabled, enabledIDs)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, nil, nil, 0, err
|
|||
|
|
}
|
|||
|
|
defer rows.Close()
|
|||
|
|
terms := make([]termRule, 0)
|
|||
|
|
dictSeen := map[string]struct{}{}
|
|||
|
|
dictNames := make([]string, 0)
|
|||
|
|
dictVersions := map[string]int{}
|
|||
|
|
var versionSum int64
|
|||
|
|
for rows.Next() {
|
|||
|
|
var term termRule
|
|||
|
|
var hint pgtype.Text
|
|||
|
|
var ref pgtype.Text
|
|||
|
|
if err := rows.Scan(
|
|||
|
|
&term.DictionaryID,
|
|||
|
|
&term.DictionaryCode,
|
|||
|
|
&term.DictionaryName,
|
|||
|
|
&term.DictionaryVersion,
|
|||
|
|
&term.PlatformScope,
|
|||
|
|
&term.ApplicablePlatforms,
|
|||
|
|
&term.MatchType,
|
|||
|
|
&term.Pattern,
|
|||
|
|
&term.Level,
|
|||
|
|
&hint,
|
|||
|
|
&ref,
|
|||
|
|
); err != nil {
|
|||
|
|
return nil, nil, nil, 0, err
|
|||
|
|
}
|
|||
|
|
term.Pattern = strings.TrimSpace(term.Pattern)
|
|||
|
|
if term.Pattern == "" {
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
if hint.Valid {
|
|||
|
|
value := hint.String
|
|||
|
|
term.Hint = &value
|
|||
|
|
}
|
|||
|
|
if ref.Valid {
|
|||
|
|
value := ref.String
|
|||
|
|
term.ReferenceLaw = &value
|
|||
|
|
}
|
|||
|
|
if term.MatchType == "regex" {
|
|||
|
|
compiled, err := regexp.Compile(term.Pattern)
|
|||
|
|
if err != nil {
|
|||
|
|
if s.logger != nil {
|
|||
|
|
s.logger.Warn("skip invalid compliance regex", zap.String("pattern", term.Pattern), zap.Error(err))
|
|||
|
|
}
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
term.compiledRegex = compiled
|
|||
|
|
term.compiledRegexPattern = term.Pattern
|
|||
|
|
}
|
|||
|
|
terms = append(terms, term)
|
|||
|
|
if _, ok := dictSeen[term.DictionaryCode]; !ok {
|
|||
|
|
dictSeen[term.DictionaryCode] = struct{}{}
|
|||
|
|
dictNames = append(dictNames, term.DictionaryName)
|
|||
|
|
dictVersions[term.DictionaryCode] = term.DictionaryVersion
|
|||
|
|
versionSum += int64(term.DictionaryVersion)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
return terms, dictNames, dictVersions, versionSum, rows.Err()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func matchViolations(policy *policySnapshot, tenantID int64, targetPlatforms []string, content string) []sharedcompliance.Violation {
|
|||
|
|
if policy == nil || !policy.MasterEnabled || policy.EnforcementMode == "disabled" {
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
content = strings.TrimSpace(content)
|
|||
|
|
if content == "" {
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
violations := make([]sharedcompliance.Violation, 0)
|
|||
|
|
for _, term := range policy.Terms {
|
|||
|
|
platforms := applicableViolationPlatforms(term, targetPlatforms)
|
|||
|
|
if platforms == nil && term.PlatformScope == "specific" {
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
switch term.MatchType {
|
|||
|
|
case "regex":
|
|||
|
|
if term.compiledRegex == nil {
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
matches := term.compiledRegex.FindAllStringIndex(content, -1)
|
|||
|
|
for _, match := range matches {
|
|||
|
|
violations = append(violations, buildViolationsForTerm(term, platforms, content, match[0], match[1])...)
|
|||
|
|
}
|
|||
|
|
case "exact":
|
|||
|
|
searchFrom := 0
|
|||
|
|
for {
|
|||
|
|
idx := strings.Index(content[searchFrom:], term.Pattern)
|
|||
|
|
if idx < 0 {
|
|||
|
|
break
|
|||
|
|
}
|
|||
|
|
start := searchFrom + idx
|
|||
|
|
end := start + len(term.Pattern)
|
|||
|
|
violations = append(violations, buildViolationsForTerm(term, platforms, content, start, end)...)
|
|||
|
|
searchFrom = end
|
|||
|
|
if searchFrom >= len(content) {
|
|||
|
|
break
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
return violations
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func applicableViolationPlatforms(term termRule, targetPlatforms []string) []string {
|
|||
|
|
if term.PlatformScope != "specific" {
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
targetSet := make(map[string]struct{}, len(targetPlatforms))
|
|||
|
|
for _, platform := range targetPlatforms {
|
|||
|
|
targetSet[platform] = struct{}{}
|
|||
|
|
}
|
|||
|
|
out := make([]string, 0)
|
|||
|
|
for _, platform := range term.ApplicablePlatforms {
|
|||
|
|
platform = strings.TrimSpace(platform)
|
|||
|
|
if platform == "" {
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
if len(targetSet) == 0 {
|
|||
|
|
out = append(out, platform)
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
if _, ok := targetSet[platform]; ok {
|
|||
|
|
out = append(out, platform)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
sort.Strings(out)
|
|||
|
|
return out
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func buildViolationsForTerm(term termRule, platforms []string, content string, start, end int) []sharedcompliance.Violation {
|
|||
|
|
matched := content[start:end]
|
|||
|
|
base := sharedcompliance.Violation{
|
|||
|
|
Source: sourceDict,
|
|||
|
|
DictionaryCode: stringPtr(term.DictionaryCode),
|
|||
|
|
DictionaryName: stringPtr(term.DictionaryName),
|
|||
|
|
TermPattern: stringPtr(term.Pattern),
|
|||
|
|
MatchedText: truncateRunes(matched, maxMatchedTextRunes),
|
|||
|
|
StartOffset: start,
|
|||
|
|
EndOffset: end,
|
|||
|
|
Level: normalizeLevel(term.Level),
|
|||
|
|
Hint: term.Hint,
|
|||
|
|
ReferenceLaw: term.ReferenceLaw,
|
|||
|
|
}
|
|||
|
|
if len(platforms) == 0 {
|
|||
|
|
return []sharedcompliance.Violation{base}
|
|||
|
|
}
|
|||
|
|
out := make([]sharedcompliance.Violation, 0, len(platforms))
|
|||
|
|
for _, platform := range platforms {
|
|||
|
|
next := base
|
|||
|
|
next.PlatformCode = stringPtr(platform)
|
|||
|
|
out = append(out, next)
|
|||
|
|
}
|
|||
|
|
return out
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) loadCheckRecord(ctx context.Context, tenantID, recordID int64) (checkRecordRow, error) {
|
|||
|
|
var record checkRecordRow
|
|||
|
|
var articleID pgtype.Int8
|
|||
|
|
var articleVersionID pgtype.Int8
|
|||
|
|
var highest pgtype.Text
|
|||
|
|
var manualReviewID pgtype.Int8
|
|||
|
|
var manualStatus string
|
|||
|
|
err := s.pool.QueryRow(ctx, `
|
|||
|
|
SELECT
|
|||
|
|
id,
|
|||
|
|
tenant_id,
|
|||
|
|
article_id,
|
|||
|
|
article_version_id,
|
|||
|
|
target_platforms,
|
|||
|
|
content_hash,
|
|||
|
|
policy_fingerprint,
|
|||
|
|
enforcement_mode,
|
|||
|
|
highest_level,
|
|||
|
|
passed,
|
|||
|
|
hit_count,
|
|||
|
|
stored_hit_count,
|
|||
|
|
truncated,
|
|||
|
|
manual_review_status,
|
|||
|
|
manual_review_id,
|
|||
|
|
checked_at,
|
|||
|
|
duration_ms
|
|||
|
|
FROM compliance_check_records
|
|||
|
|
WHERE id = $1 AND tenant_id = $2
|
|||
|
|
`, recordID, tenantID).Scan(
|
|||
|
|
&record.ID,
|
|||
|
|
&record.TenantID,
|
|||
|
|
&articleID,
|
|||
|
|
&articleVersionID,
|
|||
|
|
&record.TargetPlatforms,
|
|||
|
|
&record.ContentHash,
|
|||
|
|
&record.PolicyFingerprint,
|
|||
|
|
&record.EnforcementMode,
|
|||
|
|
&highest,
|
|||
|
|
&record.Passed,
|
|||
|
|
&record.HitCount,
|
|||
|
|
&record.StoredHitCount,
|
|||
|
|
&record.Truncated,
|
|||
|
|
&manualStatus,
|
|||
|
|
&manualReviewID,
|
|||
|
|
&record.CheckedAt,
|
|||
|
|
&record.DurationMS,
|
|||
|
|
)
|
|||
|
|
if err != nil {
|
|||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|||
|
|
return record, response.ErrNotFound(40461, "compliance_check_not_found", "compliance check record not found")
|
|||
|
|
}
|
|||
|
|
return record, response.ErrInternal(51005, "compliance_check_lookup_failed", "failed to load compliance check record")
|
|||
|
|
}
|
|||
|
|
if articleID.Valid {
|
|||
|
|
record.ArticleID = articleID.Int64
|
|||
|
|
}
|
|||
|
|
if articleVersionID.Valid {
|
|||
|
|
record.ArticleVersionID = articleVersionID.Int64
|
|||
|
|
}
|
|||
|
|
if highest.Valid {
|
|||
|
|
record.HighestLevel = &highest.String
|
|||
|
|
}
|
|||
|
|
if manualReviewID.Valid {
|
|||
|
|
record.ManualReviewID = &manualReviewID.Int64
|
|||
|
|
}
|
|||
|
|
record.ManualReviewStatus = sharedcompliance.ManualReviewStatus(manualStatus)
|
|||
|
|
return record, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) loadViolations(ctx context.Context, tenantID, recordID int64) ([]sharedcompliance.Violation, error) {
|
|||
|
|
rows, err := s.pool.Query(ctx, `
|
|||
|
|
SELECT
|
|||
|
|
id,
|
|||
|
|
record_id,
|
|||
|
|
platform_code,
|
|||
|
|
source,
|
|||
|
|
dictionary_code,
|
|||
|
|
dictionary_name,
|
|||
|
|
term_pattern,
|
|||
|
|
matched_text,
|
|||
|
|
start_offset,
|
|||
|
|
end_offset,
|
|||
|
|
dom_path,
|
|||
|
|
level,
|
|||
|
|
hint,
|
|||
|
|
reference_law
|
|||
|
|
FROM compliance_violations
|
|||
|
|
WHERE tenant_id = $1 AND record_id = $2
|
|||
|
|
ORDER BY id
|
|||
|
|
`, tenantID, recordID)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, response.ErrInternal(51008, "compliance_violation_lookup_failed", "failed to load compliance violations")
|
|||
|
|
}
|
|||
|
|
defer rows.Close()
|
|||
|
|
out := make([]sharedcompliance.Violation, 0)
|
|||
|
|
for rows.Next() {
|
|||
|
|
var v sharedcompliance.Violation
|
|||
|
|
var platform pgtype.Text
|
|||
|
|
var dictCode pgtype.Text
|
|||
|
|
var dictName pgtype.Text
|
|||
|
|
var termPattern pgtype.Text
|
|||
|
|
var domPath pgtype.Text
|
|||
|
|
var hint pgtype.Text
|
|||
|
|
var ref pgtype.Text
|
|||
|
|
if err := rows.Scan(
|
|||
|
|
&v.ID,
|
|||
|
|
&v.RecordID,
|
|||
|
|
&platform,
|
|||
|
|
&v.Source,
|
|||
|
|
&dictCode,
|
|||
|
|
&dictName,
|
|||
|
|
&termPattern,
|
|||
|
|
&v.MatchedText,
|
|||
|
|
&v.StartOffset,
|
|||
|
|
&v.EndOffset,
|
|||
|
|
&domPath,
|
|||
|
|
&v.Level,
|
|||
|
|
&hint,
|
|||
|
|
&ref,
|
|||
|
|
); err != nil {
|
|||
|
|
return nil, response.ErrInternal(51008, "compliance_violation_lookup_failed", "failed to scan compliance violations")
|
|||
|
|
}
|
|||
|
|
if platform.Valid {
|
|||
|
|
v.PlatformCode = &platform.String
|
|||
|
|
}
|
|||
|
|
if dictCode.Valid {
|
|||
|
|
v.DictionaryCode = &dictCode.String
|
|||
|
|
}
|
|||
|
|
if dictName.Valid {
|
|||
|
|
v.DictionaryName = &dictName.String
|
|||
|
|
}
|
|||
|
|
if termPattern.Valid {
|
|||
|
|
v.TermPattern = &termPattern.String
|
|||
|
|
}
|
|||
|
|
if domPath.Valid {
|
|||
|
|
v.DomPath = &domPath.String
|
|||
|
|
}
|
|||
|
|
if hint.Valid {
|
|||
|
|
v.Hint = &hint.String
|
|||
|
|
}
|
|||
|
|
if ref.Valid {
|
|||
|
|
v.ReferenceLaw = &ref.String
|
|||
|
|
}
|
|||
|
|
out = append(out, v)
|
|||
|
|
}
|
|||
|
|
return out, rows.Err()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) loadLatestManualReviewByVersion(ctx context.Context, tenantID, versionID int64) (*sharedcompliance.ManualReview, error) {
|
|||
|
|
var review sharedcompliance.ManualReview
|
|||
|
|
var checkID pgtype.Int8
|
|||
|
|
var requestNote pgtype.Text
|
|||
|
|
var reviewedBy pgtype.Int8
|
|||
|
|
var reviewedAt pgtype.Timestamptz
|
|||
|
|
var decisionReason pgtype.Text
|
|||
|
|
var cancelledBy pgtype.Int8
|
|||
|
|
var cancelledAt pgtype.Timestamptz
|
|||
|
|
var cancelReason pgtype.Text
|
|||
|
|
var summaryRaw []byte
|
|||
|
|
var platforms []string
|
|||
|
|
err := s.pool.QueryRow(ctx, `
|
|||
|
|
SELECT
|
|||
|
|
mr.id,
|
|||
|
|
mr.tenant_id,
|
|||
|
|
mr.article_id,
|
|||
|
|
mr.article_version_id,
|
|||
|
|
mr.original_check_record_id,
|
|||
|
|
mr.requested_by,
|
|||
|
|
mr.requested_at,
|
|||
|
|
mr.request_note,
|
|||
|
|
mr.status,
|
|||
|
|
mr.cancelled_by,
|
|||
|
|
mr.cancelled_at,
|
|||
|
|
mr.cancel_reason,
|
|||
|
|
mr.reviewed_by,
|
|||
|
|
mr.reviewed_at,
|
|||
|
|
mr.decision_reason,
|
|||
|
|
mr.original_violation_summary,
|
|||
|
|
mr.original_target_platforms
|
|||
|
|
FROM compliance_manual_reviews mr
|
|||
|
|
WHERE mr.tenant_id = $1
|
|||
|
|
AND mr.article_version_id = $2
|
|||
|
|
AND mr.status <> 'cancelled'
|
|||
|
|
ORDER BY mr.requested_at DESC, mr.id DESC
|
|||
|
|
LIMIT 1
|
|||
|
|
`, tenantID, versionID).Scan(
|
|||
|
|
&review.ID,
|
|||
|
|
&review.TenantID,
|
|||
|
|
&review.ArticleID,
|
|||
|
|
&review.ArticleVersionID,
|
|||
|
|
&checkID,
|
|||
|
|
&review.RequestedBy,
|
|||
|
|
&review.RequestedAt,
|
|||
|
|
&requestNote,
|
|||
|
|
&review.Status,
|
|||
|
|
&cancelledBy,
|
|||
|
|
&cancelledAt,
|
|||
|
|
&cancelReason,
|
|||
|
|
&reviewedBy,
|
|||
|
|
&reviewedAt,
|
|||
|
|
&decisionReason,
|
|||
|
|
&summaryRaw,
|
|||
|
|
&platforms,
|
|||
|
|
)
|
|||
|
|
if err != nil {
|
|||
|
|
return nil, err
|
|||
|
|
}
|
|||
|
|
if checkID.Valid {
|
|||
|
|
review.CheckRecordID = &checkID.Int64
|
|||
|
|
}
|
|||
|
|
if requestNote.Valid {
|
|||
|
|
review.RequestNote = &requestNote.String
|
|||
|
|
}
|
|||
|
|
if reviewedBy.Valid {
|
|||
|
|
review.ReviewedBy = &reviewedBy.Int64
|
|||
|
|
}
|
|||
|
|
if reviewedAt.Valid {
|
|||
|
|
review.ReviewedAt = &reviewedAt.Time
|
|||
|
|
}
|
|||
|
|
if decisionReason.Valid {
|
|||
|
|
review.DecisionReason = &decisionReason.String
|
|||
|
|
}
|
|||
|
|
if cancelledBy.Valid {
|
|||
|
|
review.CancelledBy = &cancelledBy.Int64
|
|||
|
|
}
|
|||
|
|
if cancelledAt.Valid {
|
|||
|
|
review.CancelledAt = &cancelledAt.Time
|
|||
|
|
}
|
|||
|
|
if cancelReason.Valid {
|
|||
|
|
review.CancelReason = &cancelReason.String
|
|||
|
|
}
|
|||
|
|
_ = json.Unmarshal(summaryRaw, &review.ViolationSummary)
|
|||
|
|
review.OriginalPlatforms = platforms
|
|||
|
|
return &review, nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) lookupManualReviewStatus(ctx context.Context, tenantID, versionID int64) (sharedcompliance.ManualReviewStatus, error) {
|
|||
|
|
var status string
|
|||
|
|
err := s.pool.QueryRow(ctx, `
|
|||
|
|
SELECT status
|
|||
|
|
FROM compliance_manual_reviews
|
|||
|
|
WHERE tenant_id = $1 AND article_version_id = $2 AND status <> 'cancelled'
|
|||
|
|
ORDER BY requested_at DESC, id DESC
|
|||
|
|
LIMIT 1
|
|||
|
|
`, tenantID, versionID).Scan(&status)
|
|||
|
|
if err != nil {
|
|||
|
|
return "", err
|
|||
|
|
}
|
|||
|
|
return sharedcompliance.ManualReviewStatus(status), nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func checkResultFromRecord(record checkRecordRow, violations []sharedcompliance.Violation, decision sharedcompliance.GateDecision) *sharedcompliance.CheckResult {
|
|||
|
|
return &sharedcompliance.CheckResult{
|
|||
|
|
RecordID: record.ID,
|
|||
|
|
Decision: decision,
|
|||
|
|
Passed: record.Passed,
|
|||
|
|
HighestLevel: record.HighestLevel,
|
|||
|
|
Violations: violations,
|
|||
|
|
HitCount: record.HitCount,
|
|||
|
|
StoredHitCount: record.StoredHitCount,
|
|||
|
|
Truncated: record.Truncated,
|
|||
|
|
EnforcementMode: record.EnforcementMode,
|
|||
|
|
ContentHash: record.ContentHash,
|
|||
|
|
PolicyFingerprint: record.PolicyFingerprint,
|
|||
|
|
TargetPlatforms: record.TargetPlatforms,
|
|||
|
|
ManualReviewStatus: record.ManualReviewStatus,
|
|||
|
|
ManualReviewID: record.ManualReviewID,
|
|||
|
|
CheckedAt: record.CheckedAt,
|
|||
|
|
DurationMS: record.DurationMS,
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func manualReviewOnlyResult(snapshot articleVersionSnapshot, targets []string) *sharedcompliance.CheckResult {
|
|||
|
|
decision := sharedcompliance.GateDecisionBlock
|
|||
|
|
passed := false
|
|||
|
|
if snapshot.ManualReviewStatus == sharedcompliance.ManualReviewStatusApproved {
|
|||
|
|
decision = sharedcompliance.GateDecisionPass
|
|||
|
|
passed = true
|
|||
|
|
}
|
|||
|
|
return &sharedcompliance.CheckResult{
|
|||
|
|
Decision: decision,
|
|||
|
|
Passed: passed,
|
|||
|
|
Violations: []sharedcompliance.Violation{},
|
|||
|
|
EnforcementMode: "disabled",
|
|||
|
|
ContentHash: contentHash(snapshot.Title, snapshot.Plaintext),
|
|||
|
|
PolicyFingerprint: "",
|
|||
|
|
TargetPlatforms: normalizePlatforms(targets),
|
|||
|
|
ManualReviewStatus: snapshot.ManualReviewStatus,
|
|||
|
|
ManualReviewID: snapshot.ManualReviewID,
|
|||
|
|
ManualReviewReason: snapshot.ManualReviewReason,
|
|||
|
|
ManualReviewDecidedAt: snapshot.ManualReviewDecidedAt,
|
|||
|
|
CheckedAt: time.Now().UTC(),
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func disabledCheckResult(targets []string) *sharedcompliance.CheckResult {
|
|||
|
|
return &sharedcompliance.CheckResult{
|
|||
|
|
Decision: sharedcompliance.GateDecisionPass,
|
|||
|
|
Passed: true,
|
|||
|
|
Violations: []sharedcompliance.Violation{},
|
|||
|
|
EnforcementMode: "disabled",
|
|||
|
|
TargetPlatforms: normalizePlatforms(targets),
|
|||
|
|
ManualReviewStatus: sharedcompliance.ManualReviewStatusNone,
|
|||
|
|
CheckedAt: time.Now().UTC(),
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) currentComplianceConfig() config.ComplianceConfig {
|
|||
|
|
if s != nil && s.cfg != nil {
|
|||
|
|
if cfg := s.cfg.Current(); cfg != nil {
|
|||
|
|
out := cfg.Compliance
|
|||
|
|
config.NormalizeComplianceConfig(&out)
|
|||
|
|
return out
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
out := config.ComplianceConfig{Enabled: true}
|
|||
|
|
config.NormalizeComplianceConfig(&out)
|
|||
|
|
return out
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func decide(mode string, highest *string, hitCount int) sharedcompliance.GateDecision {
|
|||
|
|
if hitCount <= 0 || highest == nil {
|
|||
|
|
return sharedcompliance.GateDecisionPass
|
|||
|
|
}
|
|||
|
|
if mode == "mandatory" {
|
|||
|
|
return sharedcompliance.GateDecisionBlock
|
|||
|
|
}
|
|||
|
|
return sharedcompliance.GateDecisionNeedsAck
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func highestViolationLevel(violations []sharedcompliance.Violation) *string {
|
|||
|
|
if len(violations) == 0 {
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
best := "info"
|
|||
|
|
for _, violation := range violations {
|
|||
|
|
if levelRank(violation.Level) > levelRank(best) {
|
|||
|
|
best = violation.Level
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
return &best
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func levelRank(level string) int {
|
|||
|
|
switch strings.TrimSpace(level) {
|
|||
|
|
case "block":
|
|||
|
|
return 4
|
|||
|
|
case "high":
|
|||
|
|
return 3
|
|||
|
|
case "medium":
|
|||
|
|
return 2
|
|||
|
|
case "info":
|
|||
|
|
return 1
|
|||
|
|
default:
|
|||
|
|
return 0
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func normalizeLevel(level string) string {
|
|||
|
|
level = strings.TrimSpace(level)
|
|||
|
|
switch level {
|
|||
|
|
case "block", "high", "medium", "info":
|
|||
|
|
return level
|
|||
|
|
default:
|
|||
|
|
return "high"
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func normalizeMode(mode string) string {
|
|||
|
|
mode = strings.TrimSpace(mode)
|
|||
|
|
switch mode {
|
|||
|
|
case "mandatory", "advisory", "disabled":
|
|||
|
|
return mode
|
|||
|
|
default:
|
|||
|
|
return "mandatory"
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func contentHash(title, plaintext string) string {
|
|||
|
|
sum := sha256.Sum256([]byte(strings.TrimSpace(title) + "\n" + strings.TrimSpace(plaintext)))
|
|||
|
|
return hex.EncodeToString(sum[:])
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func policyFingerprint(policy *policySnapshot, targets []string) string {
|
|||
|
|
targets = normalizePlatforms(targets)
|
|||
|
|
parts := []string{
|
|||
|
|
fmt.Sprintf("rev:%d", policy.Revision),
|
|||
|
|
"mode:" + policy.EnforcementMode,
|
|||
|
|
fmt.Sprintf("llm:%t", policy.LLMJudgeEnabled),
|
|||
|
|
"targets:" + strings.Join(targets, ","),
|
|||
|
|
}
|
|||
|
|
keys := make([]string, 0, len(policy.DictVersions))
|
|||
|
|
for key := range policy.DictVersions {
|
|||
|
|
keys = append(keys, key)
|
|||
|
|
}
|
|||
|
|
sort.Strings(keys)
|
|||
|
|
for _, key := range keys {
|
|||
|
|
parts = append(parts, fmt.Sprintf("%s:%d", key, policy.DictVersions[key]))
|
|||
|
|
}
|
|||
|
|
sum := sha256.Sum256([]byte(strings.Join(parts, "|")))
|
|||
|
|
return hex.EncodeToString(sum[:])
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func recordDictVersions(policy *policySnapshot) map[string]int {
|
|||
|
|
if policy == nil || policy.DictVersions == nil {
|
|||
|
|
return map[string]int{}
|
|||
|
|
}
|
|||
|
|
out := make(map[string]int, len(policy.DictVersions))
|
|||
|
|
for key, value := range policy.DictVersions {
|
|||
|
|
out[key] = value
|
|||
|
|
}
|
|||
|
|
return out
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func normalizePlatforms(platforms []string) []string {
|
|||
|
|
seen := make(map[string]struct{}, len(platforms))
|
|||
|
|
out := make([]string, 0, len(platforms))
|
|||
|
|
for _, platform := range platforms {
|
|||
|
|
platform = strings.TrimSpace(platform)
|
|||
|
|
if platform == "" {
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
if _, ok := seen[platform]; ok {
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
seen[platform] = struct{}{}
|
|||
|
|
out = append(out, platform)
|
|||
|
|
}
|
|||
|
|
sort.Strings(out)
|
|||
|
|
return out
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func decodeJSONInt64Slice(raw []byte) []int64 {
|
|||
|
|
if len(raw) == 0 {
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
var nums []int64
|
|||
|
|
if err := json.Unmarshal(raw, &nums); err == nil {
|
|||
|
|
return nums
|
|||
|
|
}
|
|||
|
|
var floats []float64
|
|||
|
|
if err := json.Unmarshal(raw, &floats); err != nil {
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
out := make([]int64, 0, len(floats))
|
|||
|
|
for _, n := range floats {
|
|||
|
|
out = append(out, int64(n))
|
|||
|
|
}
|
|||
|
|
return out
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func sameInt64Set(a, b []int64) bool {
|
|||
|
|
if len(a) != len(b) {
|
|||
|
|
return false
|
|||
|
|
}
|
|||
|
|
a = append([]int64(nil), a...)
|
|||
|
|
b = append([]int64(nil), b...)
|
|||
|
|
sort.Slice(a, func(i, j int) bool { return a[i] < a[j] })
|
|||
|
|
sort.Slice(b, func(i, j int) bool { return b[i] < b[j] })
|
|||
|
|
for i := range a {
|
|||
|
|
if a[i] != b[i] {
|
|||
|
|
return false
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
return true
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func truncateRunes(value string, maxRunes int) string {
|
|||
|
|
value = strings.TrimSpace(value)
|
|||
|
|
if maxRunes <= 0 {
|
|||
|
|
return ""
|
|||
|
|
}
|
|||
|
|
runes := []rune(value)
|
|||
|
|
if len(runes) <= maxRunes {
|
|||
|
|
return value
|
|||
|
|
}
|
|||
|
|
return string(runes[:maxRunes])
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func stringPtr(value string) *string {
|
|||
|
|
return &value
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func stringValue(value *string) string {
|
|||
|
|
if value == nil {
|
|||
|
|
return ""
|
|||
|
|
}
|
|||
|
|
return *value
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func errorString(err error) string {
|
|||
|
|
if err == nil {
|
|||
|
|
return ""
|
|||
|
|
}
|
|||
|
|
return err.Error()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func nullableString(value string) *string {
|
|||
|
|
value = strings.TrimSpace(value)
|
|||
|
|
if value == "" {
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
return &value
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func int64PtrAny(value *int64) any {
|
|||
|
|
if value == nil {
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
return *value
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func rollbackTx(tx pgx.Tx) {
|
|||
|
|
if tx == nil {
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|||
|
|
defer cancel()
|
|||
|
|
_ = tx.Rollback(ctx)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func isUniqueViolation(err error) bool {
|
|||
|
|
var pgErr *pgconn.PgError
|
|||
|
|
return errors.As(err, &pgErr) && pgErr.Code == "23505"
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) cacheManualReview(snapshot articleVersionSnapshot) {
|
|||
|
|
if snapshot.ArticleVersionID <= 0 {
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
s.reviewCache.Store(snapshot.ArticleVersionID, struct {
|
|||
|
|
snapshot articleVersionSnapshot
|
|||
|
|
expires time.Time
|
|||
|
|
}{snapshot: snapshot, expires: time.Now().Add(60 * time.Second)})
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func (s *Service) cachedManualReview(versionID int64) (articleVersionSnapshot, bool) {
|
|||
|
|
value, ok := s.reviewCache.Load(versionID)
|
|||
|
|
if !ok {
|
|||
|
|
return articleVersionSnapshot{}, false
|
|||
|
|
}
|
|||
|
|
typed, ok := value.(struct {
|
|||
|
|
snapshot articleVersionSnapshot
|
|||
|
|
expires time.Time
|
|||
|
|
})
|
|||
|
|
if !ok || time.Now().After(typed.expires) {
|
|||
|
|
s.reviewCache.Delete(versionID)
|
|||
|
|
return articleVersionSnapshot{}, false
|
|||
|
|
}
|
|||
|
|
return typed.snapshot, true
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func manualReviewGateError(snapshot articleVersionSnapshot) error {
|
|||
|
|
switch snapshot.ManualReviewStatus {
|
|||
|
|
case sharedcompliance.ManualReviewStatusPending:
|
|||
|
|
return response.ErrConflict(41007, "compliance_manual_review_pending", "current article version is waiting for manual review")
|
|||
|
|
case sharedcompliance.ManualReviewStatusRejected:
|
|||
|
|
return response.ErrConflict(41006, "compliance_manual_review_rejected", "current article version was rejected by manual review")
|
|||
|
|
default:
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
}
|