feat(ops): add job center for cross-source job operations

Provide a unified ops console for inspecting, retrying and cancelling jobs across generation, template/kol assist, knowledge parse, desktop publish/task, compliance review and monitoring collect sources. Wires RabbitMQ for retry republish and consolidates the desktop_publish_jobs columns into the base migration. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-06 12:56:39 +08:00
parent ab62d666b4
commit 7b4d7ccf68
13 changed files with 3043 additions and 12 deletions
@@ -0,0 +1,417 @@
+package app
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/geo-platform/tenant-api/internal/ops/domain"
+	"github.com/geo-platform/tenant-api/internal/ops/repository"
+	sharedconfig "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"
+)
+
+const (
+	ActionJobRetry  = "job.retry"
+	ActionJobCancel = "job.cancel"
+)
+
+type JobService struct {
+	jobs   *repository.JobRepository
+	audits *AuditService
+	mq     *rabbitmq.Client
+}
+
+func NewJobService(jobs *repository.JobRepository, audits *AuditService, mq *rabbitmq.Client) *JobService {
+	return &JobService{jobs: jobs, audits: audits, mq: mq}
+}
+
+func (s *JobService) List(ctx context.Context, filter domain.JobFilter, page, size int) (*domain.JobListResult, error) {
+	if page <= 0 {
+		page = 1
+	}
+	if size <= 0 || size > 500 {
+		size = 50
+	}
+	filter.Limit = size
+	filter.Offset = (page - 1) * size
+
+	items, total, counts, err := s.jobs.List(ctx, filter)
+	if err != nil {
+		return nil, response.ErrInternal(52101, "ops_jobs_query_failed", "failed to list jobs")
+	}
+	return &domain.JobListResult{
+		Items:  items,
+		Total:  total,
+		Page:   page,
+		Size:   size,
+		Counts: counts,
+	}, nil
+}
+
+func (s *JobService) Get(ctx context.Context, source, id string) (*domain.JobDetail, error) {
+	item, err := s.jobs.Get(ctx, source, id)
+	if err != nil {
+		return nil, mapJobError(err)
+	}
+	return item, nil
+}
+
+func (s *JobService) Retry(ctx context.Context, actor *Actor, source, id string) (*domain.JobDetail, error) {
+	before, err := s.jobs.Get(ctx, source, id)
+	if err != nil {
+		return nil, mapJobError(err)
+	}
+	if !before.CanRetry {
+		return nil, response.ErrConflict(40970, "job_retry_unavailable", "当前任务状态不支持重试")
+	}
+
+	item, err := s.jobs.Retry(ctx, source, id)
+	if err != nil {
+		return nil, mapJobError(err)
+	}
+	if err := s.publishRetry(ctx, item); err != nil {
+		_ = s.markRetryPublishFailed(context.Background(), item, err)
+		return nil, response.ErrServiceUnavailable(50370, "job_retry_queue_unavailable", "任务已回滚为失败或取消状态，队列暂不可用")
+	}
+
+	if actor != nil && s.audits != nil {
+		_ = s.audits.Append(ctx, actor.audit(ActionJobRetry, "job", 0, jobAuditMetadata(item)))
+	}
+	return item, nil
+}
+
+func (s *JobService) Cancel(ctx context.Context, actor *Actor, source, id, reason string) (*domain.JobDetail, error) {
+	before, err := s.jobs.Get(ctx, source, id)
+	if err != nil {
+		return nil, mapJobError(err)
+	}
+	if !before.CanCancel {
+		return nil, response.ErrConflict(40971, "job_cancel_unavailable", "当前任务状态不支持取消")
+	}
+
+	item, err := s.jobs.Cancel(ctx, source, id, reason)
+	if err != nil {
+		return nil, mapJobError(err)
+	}
+	if actor != nil && s.audits != nil {
+		metadata := jobAuditMetadata(item)
+		if strings.TrimSpace(reason) != "" {
+			metadata["reason"] = reason
+		}
+		_ = s.audits.Append(ctx, actor.audit(ActionJobCancel, "job", 0, metadata))
+	}
+	return item, nil
+}
+
+func (s *JobService) publishRetry(ctx context.Context, item *domain.JobDetail) error {
+	if item == nil {
+		return nil
+	}
+	switch item.Source {
+	case "generation":
+		return s.publishGenerationRetry(ctx, item)
+	case "template_assist":
+		return s.publishTemplateAssistRetry(ctx, item)
+	case "kol_assist":
+		return s.publishKolAssistRetry(ctx, item)
+	case "desktop_task":
+		return s.publishDesktopTaskRetry(ctx, item)
+	case "compliance_review":
+		return s.publishComplianceReviewRetry(ctx, item)
+	default:
+		return nil
+	}
+}
+
+func (s *JobService) markRetryPublishFailed(ctx context.Context, item *domain.JobDetail, cause error) error {
+	if item == nil || cause == nil {
+		return nil
+	}
+	message := "retry publish failed: " + cause.Error()
+	switch item.Source {
+	case "generation", "template_assist", "kol_assist", "desktop_task", "compliance_review":
+		_, err := s.jobs.Cancel(ctx, item.Source, item.ID, message)
+		return err
+	default:
+		return nil
+	}
+}
+
+func (s *JobService) publishGenerationRetry(ctx context.Context, item *domain.JobDetail) error {
+	if s.mq == nil {
+		return fmt.Errorf("rabbitmq is not configured")
+	}
+	if item.NumericID == nil || item.TenantID == nil {
+		return fmt.Errorf("generation task is missing ids")
+	}
+	body, err := json.Marshal(map[string]any{
+		"task_id":    *item.NumericID,
+		"task_type":  item.Kind,
+		"tenant_id":  *item.TenantID,
+		"article_id": int64FromMap(item.Metadata, "article_id"),
+	})
+	if err != nil {
+		return err
+	}
+	return s.mq.PublishGenerationTask(ctx, body)
+}
+
+func (s *JobService) publishTemplateAssistRetry(ctx context.Context, item *domain.JobDetail) error {
+	if s.mq == nil {
+		return fmt.Errorf("rabbitmq is not configured")
+	}
+	if item.TenantID == nil {
+		return fmt.Errorf("template assist task is missing tenant")
+	}
+	templateID := int64FromMap(item.Metadata, "template_id")
+	body, err := json.Marshal(templateAssistRetryEnvelope(item, templateID))
+	if err != nil {
+		return err
+	}
+	return s.mq.PublishTemplateAssistTask(ctx, body)
+}
+
+func (s *JobService) publishKolAssistRetry(ctx context.Context, item *domain.JobDetail) error {
+	if s.mq == nil {
+		return fmt.Errorf("rabbitmq is not configured")
+	}
+	if item.TenantID == nil {
+		return fmt.Errorf("kol assist task is missing tenant")
+	}
+	body, err := json.Marshal(map[string]any{
+		"task_id":   item.ID,
+		"tenant_id": *item.TenantID,
+	})
+	if err != nil {
+		return err
+	}
+	return s.mq.PublishKolAssistTask(ctx, body)
+}
+
+func (s *JobService) publishComplianceReviewRetry(ctx context.Context, item *domain.JobDetail) error {
+	if s.mq == nil {
+		return fmt.Errorf("rabbitmq is not configured")
+	}
+	if item.NumericID == nil || item.TenantID == nil {
+		return fmt.Errorf("compliance review job is missing ids")
+	}
+	body, err := json.Marshal(map[string]any{
+		"job_id":             *item.NumericID,
+		"message_id":         stringFromMap(item.Metadata, "message_id"),
+		"tenant_id":          *item.TenantID,
+		"article_id":         int64FromMap(item.Metadata, "article_id"),
+		"article_version_id": int64FromMap(item.Metadata, "article_version_id"),
+		"check_record_id":    int64FromMap(item.Metadata, "check_record_id"),
+	})
+	if err != nil {
+		return err
+	}
+	return s.mq.PublishComplianceReviewTask(ctx, body)
+}
+
+func (s *JobService) publishDesktopTaskRetry(ctx context.Context, item *domain.JobDetail) error {
+	if s.mq == nil {
+		return fmt.Errorf("rabbitmq is not configured")
+	}
+	if item.UUID == nil || item.WorkspaceID == nil {
+		return fmt.Errorf("desktop task is missing dispatch ids")
+	}
+	targetClientID := stringFromMap(item.Metadata, "target_client_id")
+	if targetClientID == "" {
+		return fmt.Errorf("desktop task is missing target client")
+	}
+	body, err := json.Marshal(map[string]any{
+		"type":                 "task_available",
+		"task_id":              *item.UUID,
+		"job_id":               stringFromMap(item.Metadata, "job_id"),
+		"workspace_id":         *item.WorkspaceID,
+		"target_account_id":    stringFromMap(item.Metadata, "target_account_id"),
+		"target_client_id":     targetClientID,
+		"platform":             stringFromMap(item.Metadata, "platform_id"),
+		"status":               item.Status,
+		"kind":                 item.Kind,
+		"priority":             intFromMap(item.Metadata, "priority"),
+		"lane":                 stringFromMap(item.Metadata, "lane"),
+		"interrupt_generation": intFromMap(item.Metadata, "interrupt_generation"),
+		"updated_at":           item.UpdatedAt,
+	})
+	if err != nil {
+		return err
+	}
+	if err := s.mq.PublishDesktopTaskEvent(ctx, body); err != nil {
+		return err
+	}
+	routingKey := desktopDispatchRoutingKey(s.mq.Config(), item.Kind, stringFromMap(item.Metadata, "lane"), targetClientID)
+	dispatchBody, err := json.Marshal(stream.DesktopDispatchEvent{
+		Type:                "task_available",
+		TaskID:              *item.UUID,
+		JobID:               stringFromMap(item.Metadata, "job_id"),
+		WorkspaceID:         *item.WorkspaceID,
+		TargetAccountID:     stringFromMap(item.Metadata, "target_account_id"),
+		TargetClientID:      targetClientID,
+		Platform:            stringFromMap(item.Metadata, "platform_id"),
+		Status:              item.Status,
+		Kind:                item.Kind,
+		Priority:            intFromMap(item.Metadata, "priority"),
+		Lane:                stringFromMap(item.Metadata, "lane"),
+		InterruptGeneration: intFromMap(item.Metadata, "interrupt_generation"),
+		UpdatedAt:           time.Now().UTC(),
+	})
+	if err != nil {
+		return err
+	}
+	return s.mq.PublishDesktopDispatch(ctx, routingKey, dispatchBody)
+}
+
+func templateAssistRetryEnvelope(item *domain.JobDetail, templateID int64) map[string]any {
+	templateName := ""
+	if item.Title != nil {
+		templateName = *item.Title
+	}
+	envelope := map[string]any{
+		"TaskID":       item.ID,
+		"TaskType":     item.Kind,
+		"TenantID":     int64Value(item.TenantID),
+		"TemplateID":   templateID,
+		"TemplateKey":  stringFromMap(item.Metadata, "template_key"),
+		"TemplateName": firstNonEmptyString(stringFromMap(item.Metadata, "template_name"), templateName),
+	}
+	switch item.Kind {
+	case "analyze":
+		envelope["AnalyzePrompt"] = optionalStringFromMap(item.Metadata, "analyze_prompt")
+		envelope["AnalyzeRequest"] = item.Payload
+	case "title":
+		envelope["TitlePrompt"] = optionalStringFromMap(item.Metadata, "title_prompt")
+		envelope["TitleRequest"] = item.Payload
+	case "outline":
+		envelope["OutlinePrompt"] = optionalStringFromMap(item.Metadata, "outline_prompt")
+		envelope["OutlineRequest"] = item.Payload
+	default:
+		envelope["AnalyzeRequest"] = item.Payload
+	}
+	return envelope
+}
+
+func mapJobError(err error) error {
+	if errors.Is(err, repository.ErrJobNotFound) {
+		return response.ErrNotFound(40470, "job_not_found", "任务不存在")
+	}
+	return err
+}
+
+func jobAuditMetadata(item *domain.JobDetail) map[string]any {
+	metadata := map[string]any{
+		"job_uid": item.UID,
+		"source":  item.Source,
+		"id":      item.ID,
+		"kind":    item.Kind,
+		"status":  item.Status,
+		"phase":   item.Phase,
+	}
+	if item.TenantID != nil {
+		metadata["tenant_id"] = *item.TenantID
+	}
+	if item.WorkspaceID != nil {
+		metadata["workspace_id"] = *item.WorkspaceID
+	}
+	return metadata
+}
+
+func int64Value(value *int64) int64 {
+	if value == nil {
+		return 0
+	}
+	return *value
+}
+
+func int64FromMap(values map[string]any, key string) int64 {
+	if values == nil {
+		return 0
+	}
+	switch value := values[key].(type) {
+	case int64:
+		return value
+	case int:
+		return int64(value)
+	case float64:
+		return int64(value)
+	case json.Number:
+		n, _ := value.Int64()
+		return n
+	case string:
+		n, _ := strconvParseInt64(value)
+		return n
+	default:
+		return 0
+	}
+}
+
+func intFromMap(values map[string]any, key string) int {
+	return int(int64FromMap(values, key))
+}
+
+func stringFromMap(values map[string]any, key string) string {
+	if values == nil {
+		return ""
+	}
+	switch value := values[key].(type) {
+	case string:
+		return strings.TrimSpace(value)
+	default:
+		return fmt.Sprint(value)
+	}
+}
+
+func optionalStringFromMap(values map[string]any, key string) *string {
+	value := stringFromMap(values, key)
+	if value == "" {
+		return nil
+	}
+	return &value
+}
+
+func firstNonEmptyString(values ...string) string {
+	for _, value := range values {
+		value = strings.TrimSpace(value)
+		if value != "" {
+			return value
+		}
+	}
+	return ""
+}
+
+func strconvParseInt64(value string) (int64, error) {
+	return strconv.ParseInt(strings.TrimSpace(value), 10, 64)
+}
+
+func desktopDispatchRoutingKey(cfg sharedconfig.RabbitMQConfig, kind, lane, targetClientID string) string {
+	prefix := strings.TrimSpace(cfg.DesktopDispatchPublishPrefix)
+	if strings.TrimSpace(kind) == "monitor" {
+		prefix = strings.TrimSpace(cfg.DesktopDispatchMonitorPrefix)
+	}
+	if prefix == "" {
+		if strings.TrimSpace(kind) == "monitor" {
+			prefix = "monitor"
+		} else {
+			prefix = "publish"
+		}
+	}
+	if strings.TrimSpace(kind) == "monitor" {
+		laneSegment := "normal"
+		switch strings.TrimSpace(lane) {
+		case "high":
+			laneSegment = "high"
+		case "retry":
+			laneSegment = "retry"
+		}
+		return fmt.Sprintf("%s.%s.%s", prefix, laneSegment, targetClientID)
+	}
+	return fmt.Sprintf("%s.%s", prefix, targetClientID)
+}