root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(compliance): add content compliance detection across tenant, ops, and clients

Browse Source 
Implements the Revision 8/9 compliance design: tenant + ops backends, ops-web
content-safety pages, admin-web editor/publish gate integration, desktop publish
block surfacing, and supporting migrations / shared types / config plumbing.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
liangxu
2026-05-05 20:48:14 +08:00
parent 81577b6154
commit 745cdd79cf
73 changed files with 12747 additions and 1892 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/tenant/app/desktop_task_service.go
+177 -37
View File
@@ -16,9 +16,12 @@ import (
"github.com/geo-platform/tenant-api/internal/shared/auth"
sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache"
sharedcompliance "github.com/geo-platform/tenant-api/internal/shared/compliance"
"github.com/geo-platform/tenant-api/internal/shared/config"
"github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq"
"github.com/geo-platform/tenant-api/internal/shared/response"
"github.com/geo-platform/tenant-api/internal/shared/stream"
tenantcompliance "github.com/geo-platform/tenant-api/internal/tenant/compliance"
"github.com/geo-platform/tenant-api/internal/tenant/repository"
)
@@ -30,6 +33,7 @@ type DesktopTaskService struct {
redis *goredis.Client
messaging *rabbitmq.Client
logger *zap.Logger
compliance *tenantcompliance.Service
}
func NewDesktopTaskService(pool *pgxpool.Pool, monitoringPool *pgxpool.Pool, messaging *rabbitmq.Client, logger *zap.Logger) *DesktopTaskService {
@@ -39,9 +43,18 @@ func NewDesktopTaskService(pool *pgxpool.Pool, monitoringPool *pgxpool.Pool, mes
repo: repository.NewDesktopTaskRepository(pool),
messaging: messaging,
logger: logger,
compliance: tenantcompliance.NewService(pool, config.NewStaticProvider(&config.Config{
Compliance: config.ComplianceConfig{Enabled: true},
}), logger).WithRabbitMQ(messaging),
}
}
func NewDesktopTaskServiceWithConfig(pool *pgxpool.Pool, monitoringPool *pgxpool.Pool, messaging *rabbitmq.Client, logger *zap.Logger, cfg config.Provider) *DesktopTaskService {
svc := NewDesktopTaskService(pool, monitoringPool, messaging, logger)
svc.compliance = tenantcompliance.NewService(pool, cfg, logger).WithRabbitMQ(messaging)
return svc
}
func (s *DesktopTaskService) WithCache(c sharedcache.Cache) *DesktopTaskService {
s.cache = c
return s
@@ -55,24 +68,28 @@ func (s *DesktopTaskService) WithRedis(redis *goredis.Client) *DesktopTaskServic
}
type DesktopTaskView struct {
ID string `json:"id"`
JobID string `json:"job_id"`
TenantID int64 `json:"tenant_id"`
WorkspaceID int64 `json:"workspace_id"`
TargetAccountID string `json:"target_account_id"`
TargetClientID string `json:"target_client_id"`
Platform string `json:"platform"`
Kind string `json:"kind"`
Payload json.RawMessage `json:"payload"`
Status string `json:"status"`
DedupKey *string `json:"dedup_key"`
ActiveAttemptID *string `json:"active_attempt_id"`
LeaseExpiresAt *time.Time `json:"lease_expires_at"`
Attempts int `json:"attempts"`
Result json.RawMessage `json:"result"`
Error json.RawMessage `json:"error"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
ID string `json:"id"`
JobID string `json:"job_id"`
TenantID int64 `json:"tenant_id"`
WorkspaceID int64 `json:"workspace_id"`
TargetAccountID string `json:"target_account_id"`
TargetClientID string `json:"target_client_id"`
Platform string `json:"platform"`
Kind string `json:"kind"`
Payload json.RawMessage `json:"payload"`
Status string `json:"status"`
PublishJobStatus *string `json:"publish_job_status,omitempty"`
ComplianceBlockedRecordID *int64 `json:"compliance_blocked_record_id,omitempty"`
ComplianceBlockedAt *time.Time `json:"compliance_blocked_at,omitempty"`
ComplianceBlockedReason *string `json:"compliance_blocked_reason,omitempty"`
DedupKey *string `json:"dedup_key"`
ActiveAttemptID *string `json:"active_attempt_id"`
LeaseExpiresAt *time.Time `json:"lease_expires_at"`
Attempts int `json:"attempts"`
Result json.RawMessage `json:"result"`
Error json.RawMessage `json:"error"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
type LeaseDesktopTaskRequest struct {
@@ -144,6 +161,9 @@ func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.Deskt
if err := s.recoverExpiredClientTasks(ctx, client); err != nil {
return nil, err
}
if err := s.recheckQueuedPublishTasksForClient(ctx, client); err != nil {
return nil, err
}
rawToken, tokenHash, err := newDesktopClientToken()
if err != nil {
@@ -230,6 +250,86 @@ func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.Deskt
}, nil
}
func (s *DesktopTaskService) recheckQueuedPublishTasksForClient(ctx context.Context, client *repository.DesktopClient) error {
if s == nil || s.pool == nil || s.compliance == nil || client == nil {
return nil
}
rows, err := s.pool.Query(ctx, `
SELECT DISTINCT
j.desktop_id,
j.tenant_id,
j.workspace_id,
j.created_by_user_id,
j.article_id,
j.article_version_id,
COALESCE(array_agg(DISTINCT dt.platform_id) FILTER (WHERE dt.platform_id <> ''), '{}') AS platforms
FROM desktop_tasks dt
JOIN desktop_publish_jobs j ON j.desktop_id = dt.job_id
WHERE dt.target_client_id = $1
AND dt.kind = 'publish'
AND dt.status = 'queued'
AND j.status = 'queued'
AND j.article_id IS NOT NULL
AND j.article_version_id IS NOT NULL
GROUP BY j.desktop_id, j.tenant_id, j.workspace_id, j.created_by_user_id, j.article_id, j.article_version_id
ORDER BY MIN(dt.created_at)
LIMIT 5
`, client.ID)
if err != nil {
return response.ErrInternal(50118, "desktop_publish_compliance_recheck_failed", "failed to load publish jobs for compliance recheck")
}
defer rows.Close()
type candidate struct {
JobID uuid.UUID
TenantID int64
WorkspaceID int64
CreatedByUserID int64
ArticleID int64
ArticleVersionID int64
Platforms []string
}
candidates := make([]candidate, 0)
for rows.Next() {
var item candidate
if scanErr := rows.Scan(&item.JobID, &item.TenantID, &item.WorkspaceID, &item.CreatedByUserID, &item.ArticleID, &item.ArticleVersionID, &item.Platforms); scanErr != nil {
return response.ErrInternal(50118, "desktop_publish_compliance_recheck_failed", "failed to scan publish job for compliance recheck")
}
if item.WorkspaceID == client.WorkspaceID && item.TenantID == client.TenantID {
candidates = append(candidates, item)
}
}
if err := rows.Err(); err != nil {
return response.ErrInternal(50118, "desktop_publish_compliance_recheck_failed", "failed to iterate publish jobs for compliance recheck")
}
for _, item := range candidates {
gate, gateErr := s.compliance.GateForPublish(ctx, tenantcompliance.GateForPublishRequest{
TenantID: item.TenantID,
ArticleID: item.ArticleID,
ArticleVersionID: item.ArticleVersionID,
ActorID: item.CreatedByUserID,
TargetPlatforms: item.Platforms,
TriggerSource: "scheduler_recheck",
})
if gateErr != nil && gate != nil && gate.Decision == sharedcompliance.GateDecisionBlock {
if markErr := s.compliance.MarkPublishJobBlockedByCompliance(ctx, item.JobID, gate.Result); markErr != nil {
return markErr
}
continue
}
if gateErr != nil {
return response.ErrServiceUnavailable(41004, "compliance_engine_unavailable", "scheduled publish compliance recheck failed")
}
if gate != nil && gate.Decision == sharedcompliance.GateDecisionBlock {
if markErr := s.compliance.MarkPublishJobBlockedByCompliance(ctx, item.JobID, gate.Result); markErr != nil {
return markErr
}
}
}
return nil
}
const desktopTaskRepositoryReturningColumns = `
t.desktop_id,
t.job_id,
@@ -862,7 +962,11 @@ func (s *DesktopTaskService) listPublishTasksByStatuses(
t.result,
t.error,
t.created_at,
t.updated_at
t.updated_at,
j.status,
j.compliance_blocked_record_id,
j.compliance_blocked_at,
j.compliance_blocked_reason
FROM desktop_tasks AS t
JOIN desktop_publish_jobs AS j
ON j.desktop_id = t.job_id
@@ -945,6 +1049,10 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) {
errorJSON []byte
createdAt time.Time
updatedAt time.Time
jobStatus string
blockedRecordID pgtype.Int8
blockedAt pgtype.Timestamptz
blockedReason pgtype.Text
)
if err := rows.Scan(
&desktopID,
@@ -965,6 +1073,10 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) {
&errorJSON,
&createdAt,
&updatedAt,
&jobStatus,
&blockedRecordID,
&blockedAt,
&blockedReason,
); err != nil {
return nil, response.ErrInternal(50110, "desktop_publish_tasks_scan_failed", "failed to scan desktop publish tasks")
}
@@ -990,25 +1102,53 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) {
leaseExpiresAtValue = &value
}
publishJobStatusText := strings.TrimSpace(jobStatus)
var publishJobStatus *string
if publishJobStatusText != "" {
publishJobStatus = &publishJobStatusText
}
var complianceBlockedRecordID *int64
if blockedRecordID.Valid {
value := blockedRecordID.Int64
complianceBlockedRecordID = &value
}
var complianceBlockedAt *time.Time
if blockedAt.Valid {
value := blockedAt.Time
complianceBlockedAt = &value
}
var complianceBlockedReason *string
if blockedReason.Valid {
value := blockedReason.String
complianceBlockedReason = &value
}
items = append(items, DesktopTaskView{
ID: desktopID.String(),
JobID: jobID.String(),
TenantID: tenantID,
WorkspaceID: workspaceID,
TargetAccountID: targetAccountID.String(),
TargetClientID: targetClientID.String(),
Platform: platform,
Kind: kind,
Payload: json.RawMessage(payload),
Status: normalizeDesktopTaskTerminalStatus(status),
DedupKey: dedupKeyText,
ActiveAttemptID: activeAttemptIDText,
LeaseExpiresAt: leaseExpiresAtValue,
Attempts: int(attempts),
Result: json.RawMessage(resultJSON),
Error: json.RawMessage(errorJSON),
CreatedAt: createdAt,
UpdatedAt: updatedAt,
ID: desktopID.String(),
JobID: jobID.String(),
TenantID: tenantID,
WorkspaceID: workspaceID,
TargetAccountID: targetAccountID.String(),
TargetClientID: targetClientID.String(),
Platform: platform,
Kind: kind,
Payload: json.RawMessage(payload),
Status: normalizeDesktopTaskTerminalStatus(status),
PublishJobStatus: publishJobStatus,
ComplianceBlockedRecordID: complianceBlockedRecordID,
ComplianceBlockedAt: complianceBlockedAt,
ComplianceBlockedReason: complianceBlockedReason,
DedupKey: dedupKeyText,
ActiveAttemptID: activeAttemptIDText,
LeaseExpiresAt: leaseExpiresAtValue,
Attempts: int(attempts),
Result: json.RawMessage(resultJSON),
Error: json.RawMessage(errorJSON),
CreatedAt: createdAt,
UpdatedAt: updatedAt,
})
}
if err := rows.Err(); err != nil {