diff --git a/apps/ops-web/src/layouts/AppShell.vue b/apps/ops-web/src/layouts/AppShell.vue index c204ca2..fb35cae 100644 --- a/apps/ops-web/src/layouts/AppShell.vue +++ b/apps/ops-web/src/layouts/AppShell.vue @@ -53,6 +53,7 @@ import { CrownOutlined, FileSearchOutlined, GlobalOutlined, + PartitionOutlined, LineChartOutlined, SafetyCertificateOutlined, SettingOutlined, @@ -87,6 +88,7 @@ const menuLeaves: MenuLeaf[] = [ { key: '/compliance/records', label: '检测记录', path: '/compliance/records' }, { key: '/compliance/stats', label: '合规统计', path: '/compliance/stats' }, { key: '/accounts', label: '操作员管理', path: '/accounts' }, + { key: '/jobs', label: '任务中心', path: '/jobs' }, { key: '/audits', label: '审计日志', path: '/audits' }, ] @@ -148,6 +150,11 @@ const menuItems = computed(() => [ label: '操作员管理', icon: () => h(SafetyCertificateOutlined), }, + { + key: '/jobs', + label: '任务中心', + icon: () => h(PartitionOutlined), + }, { key: '/audits', label: '审计日志', diff --git a/apps/ops-web/src/router/index.ts b/apps/ops-web/src/router/index.ts index d1e070c..48b21f3 100644 --- a/apps/ops-web/src/router/index.ts +++ b/apps/ops-web/src/router/index.ts @@ -42,6 +42,12 @@ export const router = createRouter({ component: () => import('@/views/AuditLogsView.vue'), meta: { title: '审计日志' }, }, + { + path: 'jobs', + name: 'jobs', + component: () => import('@/views/JobsView.vue'), + meta: { title: '任务中心' }, + }, { path: 'site-domain-mappings', name: 'site-domain-mappings', diff --git a/apps/ops-web/src/views/JobsView.vue b/apps/ops-web/src/views/JobsView.vue new file mode 100644 index 0000000..6d004a4 --- /dev/null +++ b/apps/ops-web/src/views/JobsView.vue @@ -0,0 +1,574 @@ + + + + + diff --git a/server/cmd/ops-api/main.go b/server/cmd/ops-api/main.go index cf9e2de..9f7377b 100644 --- a/server/cmd/ops-api/main.go +++ b/server/cmd/ops-api/main.go @@ -19,6 +19,7 @@ import ( sharedauth "github.com/geo-platform/tenant-api/internal/shared/auth" sharedbootstrap "github.com/geo-platform/tenant-api/internal/shared/bootstrap" "github.com/geo-platform/tenant-api/internal/shared/cache" + "github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq" "github.com/geo-platform/tenant-api/internal/shared/observability" "github.com/geo-platform/tenant-api/internal/shared/repository/postgres" "github.com/geo-platform/tenant-api/internal/shared/repository/redis" @@ -64,8 +65,20 @@ func main() { defer func() { _ = rdb.Close() }() appCache := cache.New(cfg.Cache.Driver, rdb) + var rabbitMQ *rabbitmq.Client + if cfg.RabbitMQ.URL != "" { + mq, mqErr := rabbitmq.New(ctx, cfg.RabbitMQ) + if mqErr != nil { + logger.Warn("rabbitmq unavailable during startup; job retry actions requiring queues are disabled", zap.Error(mqErr)) + } else { + rabbitMQ = mq + defer func() { _ = rabbitMQ.Close() }() + } + } + accountsRepo := repository.NewAccountRepository(pool) adminUsersRepo := repository.NewAdminUserRepository(pool) + jobsRepo := repository.NewJobRepository(pool, monitoringPool) kolSubscriptionsRepo := repository.NewKolSubscriptionRepository(pool) auditsRepo := repository.NewAuditRepository(pool) siteDomainMappingsRepo := repository.NewSiteDomainMappingRepository(monitoringPool) @@ -83,6 +96,7 @@ func main() { accountSvc := app.NewAccountService(accountsRepo, auditSvc) tenantLoginGuard := sharedauth.NewLoginGuard(rdb, sharedauth.DefaultLoginGuardConfig("tenant")) adminUserSvc := app.NewAdminUserService(adminUsersRepo, auditSvc, cfg.AdminUsers.DefaultPlanCode).WithLoginGuard(tenantLoginGuard) + jobSvc := app.NewJobService(jobsRepo, auditSvc, rabbitMQ) kolSubscriptionSvc := app.NewKolSubscriptionService(kolSubscriptionsRepo, auditSvc).WithCache(appCache) siteDomainMappingSvc := app.NewSiteDomainMappingService(siteDomainMappingsRepo, auditSvc) complianceSvc := app.NewComplianceService(pool, auditSvc, logger) @@ -127,6 +141,7 @@ func main() { Auth: authSvc, Accounts: accountSvc, AdminUsers: adminUserSvc, + Jobs: jobSvc, KolSubs: kolSubscriptionSvc, Audits: auditSvc, SiteDomains: siteDomainMappingSvc, diff --git a/server/internal/ops/app/job.go b/server/internal/ops/app/job.go new file mode 100644 index 0000000..631f084 --- /dev/null +++ b/server/internal/ops/app/job.go @@ -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) +} diff --git a/server/internal/ops/config/config.go b/server/internal/ops/config/config.go index f90e78b..e9129ab 100644 --- a/server/internal/ops/config/config.go +++ b/server/internal/ops/config/config.go @@ -25,6 +25,7 @@ type Config struct { MonitoringDatabase sharedconfig.DatabaseConfig `mapstructure:"monitoring_database"` Redis sharedconfig.RedisConfig `mapstructure:"redis"` Cache sharedconfig.CacheConfig `mapstructure:"cache"` + RabbitMQ sharedconfig.RabbitMQConfig `mapstructure:"rabbitmq"` Log sharedconfig.LogConfig `mapstructure:"log"` JWT JWTConfig `mapstructure:"jwt"` DefaultAdmin DefaultAdminConfig `mapstructure:"default_admin"` @@ -232,6 +233,9 @@ func Diff(previous, current *Config) []FieldChange { if previous.Cache != current.Cache { add("cache", false) } + if previous.RabbitMQ != current.RabbitMQ { + add("rabbitmq", false) + } if previous.Log != current.Log { add("log", false) } @@ -340,6 +344,7 @@ func normalizeConfig(cfg *Config) { cfg.Cache.Driver = "redis" } sharedconfig.NormalizeCacheConfig(&cfg.Cache) + normalizeOpsRabbitMQConfig(&cfg.RabbitMQ) if strings.TrimSpace(cfg.Log.Level) == "" { cfg.Log.Level = "info" } @@ -374,6 +379,7 @@ func defaultSettings() map[string]any { "cache": map[string]any{ "driver": "redis", }, + "rabbitmq": map[string]any{}, "log": map[string]any{ "level": "info", "format": "json", @@ -423,9 +429,57 @@ func applyEnvOverrides(cfg *Config) error { if v := os.Getenv("OPS_SERVER_TRUSTED_PROXIES"); v != "" { cfg.Server.TrustedProxies = strings.Split(v, ",") } + if v := os.Getenv("OPS_RABBITMQ_URL"); v != "" { + cfg.RabbitMQ.URL = v + } return nil } +func normalizeOpsRabbitMQConfig(cfg *sharedconfig.RabbitMQConfig) { + if cfg == nil { + return + } + if strings.TrimSpace(cfg.GenerationExchange) == "" { + cfg.GenerationExchange = "generation.task" + } + if strings.TrimSpace(cfg.GenerationRoutingKey) == "" { + cfg.GenerationRoutingKey = "generation.task.run" + } + if strings.TrimSpace(cfg.DesktopTaskEventExchange) == "" { + cfg.DesktopTaskEventExchange = "desktop.task.event" + } + if strings.TrimSpace(cfg.DesktopDispatchExchange) == "" { + cfg.DesktopDispatchExchange = "desktop.task.dispatch" + } + if strings.TrimSpace(cfg.DesktopDispatchPublishPrefix) == "" { + cfg.DesktopDispatchPublishPrefix = "publish" + } + if strings.TrimSpace(cfg.DesktopDispatchMonitorPrefix) == "" { + cfg.DesktopDispatchMonitorPrefix = "monitor" + } + if strings.TrimSpace(cfg.TemplateAssistExchange) == "" { + cfg.TemplateAssistExchange = "template.assist" + } + if strings.TrimSpace(cfg.TemplateAssistRoutingKey) == "" { + cfg.TemplateAssistRoutingKey = "template.assist.run" + } + if strings.TrimSpace(cfg.KolAssistExchange) == "" { + cfg.KolAssistExchange = "kol.assist" + } + if strings.TrimSpace(cfg.KolAssistRoutingKey) == "" { + cfg.KolAssistRoutingKey = "kol.assist.run" + } + if strings.TrimSpace(cfg.ComplianceReviewExchange) == "" { + cfg.ComplianceReviewExchange = "compliance.review" + } + if strings.TrimSpace(cfg.ComplianceReviewRoutingKey) == "" { + cfg.ComplianceReviewRoutingKey = "compliance.review.run" + } + if cfg.PublishChannelPoolSize <= 0 { + cfg.PublishChannelPoolSize = 16 + } +} + type fileSnapshot struct { size int64 modTime time.Time diff --git a/server/internal/ops/domain/job.go b/server/internal/ops/domain/job.go new file mode 100644 index 0000000..455f97d --- /dev/null +++ b/server/internal/ops/domain/job.go @@ -0,0 +1,85 @@ +package domain + +import "time" + +const ( + JobPhaseQueued = "queued" + JobPhaseRunning = "running" + JobPhaseSucceeded = "succeeded" + JobPhaseFailed = "failed" + JobPhaseCanceled = "canceled" + JobPhaseBlocked = "blocked" + JobPhaseUnknown = "unknown" +) + +type JobFilter struct { + Source string + Phase string + Status string + Kind string + Keyword string + TenantID *int64 + WorkspaceID *int64 + UserID *int64 + StartAt *time.Time + EndAt *time.Time + Limit int + Offset int +} + +type JobSummary struct { + UID string `json:"uid"` + Source string `json:"source"` + ID string `json:"id"` + NumericID *int64 `json:"numeric_id,omitempty"` + UUID *string `json:"uuid,omitempty"` + Kind string `json:"kind"` + Status string `json:"status"` + Phase string `json:"phase"` + TenantID *int64 `json:"tenant_id,omitempty"` + TenantName *string `json:"tenant_name,omitempty"` + WorkspaceID *int64 `json:"workspace_id,omitempty"` + WorkspaceName *string `json:"workspace_name,omitempty"` + UserID *int64 `json:"user_id,omitempty"` + UserName *string `json:"user_name,omitempty"` + Title *string `json:"title,omitempty"` + Target *string `json:"target,omitempty"` + Queue *string `json:"queue,omitempty"` + Worker *string `json:"worker,omitempty"` + Attempts *int `json:"attempts,omitempty"` + Priority *int `json:"priority,omitempty"` + ErrorMessage *string `json:"error_message,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + StartedAt *time.Time `json:"started_at,omitempty"` + CompletedAt *time.Time `json:"completed_at,omitempty"` + LeaseExpiresAt *time.Time `json:"lease_expires_at,omitempty"` + AvailableAt *time.Time `json:"available_at,omitempty"` + LastHeartbeatAt *time.Time `json:"last_heartbeat_at,omitempty"` + DurationSeconds *int64 `json:"duration_seconds,omitempty"` + IsStuck bool `json:"is_stuck"` + CanRetry bool `json:"can_retry"` + CanCancel bool `json:"can_cancel"` +} + +type JobCounts struct { + Total int64 `json:"total"` + ByPhase map[string]int64 `json:"by_phase"` + BySource map[string]int64 `json:"by_source"` +} + +type JobListResult struct { + Items []JobSummary `json:"items"` + Total int64 `json:"total"` + Page int `json:"page"` + Size int `json:"size"` + Counts JobCounts `json:"counts"` +} + +type JobDetail struct { + JobSummary + Payload any `json:"payload,omitempty"` + Result any `json:"result,omitempty"` + Error any `json:"error,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} diff --git a/server/internal/ops/repository/job.go b/server/internal/ops/repository/job.go new file mode 100644 index 0000000..d3a85bb --- /dev/null +++ b/server/internal/ops/repository/job.go @@ -0,0 +1,1745 @@ +package repository + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strconv" + "strings" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/geo-platform/tenant-api/internal/ops/domain" +) + +var ErrJobNotFound = errors.New("job not found") + +type JobRepository struct { + pool *pgxpool.Pool + monitoringPool *pgxpool.Pool +} + +func NewJobRepository(pool, monitoringPool *pgxpool.Pool) *JobRepository { + return &JobRepository{pool: pool, monitoringPool: monitoringPool} +} + +func (r *JobRepository) List(ctx context.Context, f domain.JobFilter) ([]domain.JobSummary, int64, domain.JobCounts, error) { + limit := f.Limit + if limit <= 0 || limit > 500 { + limit = 50 + } + offset := f.Offset + if offset < 0 { + offset = 0 + } + + businessItems, businessErr := r.listBusinessJobs(ctx, f) + if businessErr != nil { + return nil, 0, domain.JobCounts{}, businessErr + } + monitoringItems, monitoringErr := r.listMonitoringJobs(ctx, f) + if monitoringErr != nil { + return nil, 0, domain.JobCounts{}, monitoringErr + } + + all := append(businessItems, monitoringItems...) + sortJobs(all) + counts := buildJobCounts(all) + + total := int64(len(all)) + if offset >= len(all) { + return []domain.JobSummary{}, total, counts, nil + } + end := offset + limit + if end > len(all) { + end = len(all) + } + return all[offset:end], total, counts, nil +} + +func (r *JobRepository) Get(ctx context.Context, source, id string) (*domain.JobDetail, error) { + source = normalizeJobSource(source) + id = strings.TrimSpace(id) + switch source { + case "generation": + return r.getGenerationJob(ctx, id) + case "template_assist": + return r.getTemplateAssistJob(ctx, id) + case "kol_assist": + return r.getKolAssistJob(ctx, id) + case "knowledge_parse": + return r.getKnowledgeParseJob(ctx, id) + case "desktop_publish": + return r.getDesktopPublishJob(ctx, id) + case "desktop_task": + return r.getDesktopTaskJob(ctx, id) + case "compliance_review": + return r.getComplianceReviewJob(ctx, id) + case "monitoring_collect": + return r.getMonitoringCollectJob(ctx, id) + default: + return nil, ErrJobNotFound + } +} + +func (r *JobRepository) Retry(ctx context.Context, source, id string) (*domain.JobDetail, error) { + source = normalizeJobSource(source) + switch source { + case "generation": + return r.retryGenerationJob(ctx, id) + case "template_assist": + return r.retryTemplateAssistJob(ctx, id) + case "kol_assist": + return r.retryKolAssistJob(ctx, id) + case "knowledge_parse": + return r.retryKnowledgeParseJob(ctx, id) + case "desktop_task": + return r.retryDesktopTaskJob(ctx, id) + case "compliance_review": + return r.retryComplianceReviewJob(ctx, id) + case "monitoring_collect": + return r.retryMonitoringCollectJob(ctx, id) + default: + return nil, ErrJobNotFound + } +} + +func (r *JobRepository) Cancel(ctx context.Context, source, id, reason string) (*domain.JobDetail, error) { + source = normalizeJobSource(source) + reason = strings.TrimSpace(reason) + if reason == "" { + reason = "cancelled from ops job console" + } + switch source { + case "generation": + return r.cancelGenerationJob(ctx, id, reason) + case "template_assist": + return r.cancelTemplateAssistJob(ctx, id, reason) + case "kol_assist": + return r.cancelKolAssistJob(ctx, id, reason) + case "knowledge_parse": + return r.cancelKnowledgeParseJob(ctx, id, reason) + case "desktop_publish": + return r.cancelDesktopPublishJob(ctx, id, reason) + case "desktop_task": + return r.cancelDesktopTaskJob(ctx, id, reason) + case "compliance_review": + return r.cancelComplianceReviewJob(ctx, id, reason) + case "monitoring_collect": + return r.cancelMonitoringCollectJob(ctx, id, reason) + default: + return nil, ErrJobNotFound + } +} + +func (r *JobRepository) listBusinessJobs(ctx context.Context, f domain.JobFilter) ([]domain.JobSummary, error) { + if r == nil || r.pool == nil { + return nil, nil + } + args := make([]any, 0) + sourceFilter := normalizeJobSource(f.Source) + sourcePredicate := "TRUE" + if sourceFilter != "" { + args = append(args, sourceFilter) + sourcePredicate = fmt.Sprintf("source = $%d", len(args)) + } + sourceArg := len(args) + 1 + args = append(args, nullableTextValue(sourceFilter)) + phaseArg := len(args) + 1 + args = append(args, nullableTextValue(f.Phase)) + statusArg := len(args) + 1 + args = append(args, nullableTextValue(f.Status)) + kindArg := len(args) + 1 + args = append(args, nullableTextValue(f.Kind)) + keywordArg := len(args) + 1 + args = append(args, nullableKeyword(f.Keyword)) + tenantArg := len(args) + 1 + args = append(args, f.TenantID) + workspaceArg := len(args) + 1 + args = append(args, f.WorkspaceID) + userArg := len(args) + 1 + args = append(args, f.UserID) + startArg := len(args) + 1 + args = append(args, f.StartAt) + endArg := len(args) + 1 + args = append(args, f.EndAt) + + rows, err := r.pool.Query(ctx, fmt.Sprintf(` +WITH jobs AS ( + %s +) +SELECT + source, id, numeric_id, uuid, kind, status, phase, + tenant_id, tenant_name, workspace_id, workspace_name, user_id, user_name, + title, target, queue, worker, attempts, priority, error_message, + created_at, updated_at, started_at, completed_at, lease_expires_at, + available_at, last_heartbeat_at +FROM jobs +WHERE %s + AND ($%d::text IS NULL OR source = $%d) + AND ($%d::text IS NULL OR phase = $%d) + AND ($%d::text IS NULL OR status = $%d) + AND ($%d::text IS NULL OR kind ILIKE '%%' || $%d || '%%') + AND ($%d::text IS NULL OR + id ILIKE '%%' || $%d || '%%' OR + COALESCE(uuid, '') ILIKE '%%' || $%d || '%%' OR + COALESCE(title, '') ILIKE '%%' || $%d || '%%' OR + COALESCE(target, '') ILIKE '%%' || $%d || '%%' OR + COALESCE(error_message, '') ILIKE '%%' || $%d || '%%' OR + COALESCE(tenant_name, '') ILIKE '%%' || $%d || '%%' OR + COALESCE(workspace_name, '') ILIKE '%%' || $%d || '%%' OR + COALESCE(user_name, '') ILIKE '%%' || $%d || '%%') + AND ($%d::bigint IS NULL OR tenant_id = $%d) + AND ($%d::bigint IS NULL OR workspace_id = $%d) + AND ($%d::bigint IS NULL OR user_id = $%d) + AND ($%d::timestamptz IS NULL OR created_at >= $%d) + AND ($%d::timestamptz IS NULL OR created_at < $%d) +ORDER BY created_at DESC, source ASC, id DESC +LIMIT 2000`, + businessJobUnionSQL(), + sourcePredicate, + sourceArg, sourceArg, + phaseArg, phaseArg, + statusArg, statusArg, + kindArg, kindArg, + keywordArg, keywordArg, keywordArg, keywordArg, keywordArg, keywordArg, keywordArg, keywordArg, keywordArg, + tenantArg, tenantArg, + workspaceArg, workspaceArg, + userArg, userArg, + startArg, startArg, + endArg, endArg, + ), args...) + if err != nil { + return nil, err + } + defer rows.Close() + + items := make([]domain.JobSummary, 0) + for rows.Next() { + item, err := scanJobSummary(rows) + if err != nil { + return nil, err + } + items = append(items, *item) + } + return items, rows.Err() +} + +func (r *JobRepository) listMonitoringJobs(ctx context.Context, f domain.JobFilter) ([]domain.JobSummary, error) { + if r == nil || r.monitoringPool == nil { + return nil, nil + } + sourceFilter := normalizeJobSource(f.Source) + if sourceFilter != "" && sourceFilter != "monitoring_collect" { + return nil, nil + } + + args := make([]any, 0) + phaseArg := len(args) + 1 + args = append(args, nullableTextValue(f.Phase)) + statusArg := len(args) + 1 + args = append(args, nullableTextValue(f.Status)) + kindArg := len(args) + 1 + args = append(args, nullableTextValue(f.Kind)) + keywordArg := len(args) + 1 + args = append(args, nullableKeyword(f.Keyword)) + tenantArg := len(args) + 1 + args = append(args, f.TenantID) + workspaceArg := len(args) + 1 + args = append(args, f.WorkspaceID) + userArg := len(args) + 1 + args = append(args, f.UserID) + startArg := len(args) + 1 + args = append(args, f.StartAt) + endArg := len(args) + 1 + args = append(args, f.EndAt) + + rows, err := r.monitoringPool.Query(ctx, fmt.Sprintf(` +WITH jobs AS ( + SELECT + 'monitoring_collect'::text AS source, + t.id::text AS id, + t.id::bigint AS numeric_id, + NULL::text AS uuid, + ('monitor:' || t.ai_platform_id || ':' || t.run_mode)::text AS kind, + t.status::text AS status, + CASE + WHEN t.status IN ('pending', 'accepted', 'dispatching') THEN 'queued' + WHEN t.status IN ('leased', 'running', 'received') THEN 'running' + WHEN t.status IN ('completed', 'succeeded') THEN 'succeeded' + WHEN t.status IN ('failed', 'expired') THEN 'failed' + WHEN t.status IN ('skipped', 'superseded') THEN 'canceled' + ELSE 'unknown' + END AS phase, + t.tenant_id::bigint AS tenant_id, + NULL::text AS tenant_name, + r.workspace_id::bigint AS workspace_id, + NULL::text AS workspace_name, + r.requested_by_user_id::bigint AS user_id, + NULL::text AS user_name, + ('Brand #' || t.brand_id || ' / Question #' || t.question_id)::text AS title, + (t.ai_platform_id || ' / ' || t.business_date::text)::text AS target, + ('monitoring:' || COALESCE(t.dispatch_lane, 'normal'))::text AS queue, + COALESCE(t.leased_to_executor, t.target_client_id::text)::text AS worker, + COALESCE(t.dispatch_attempts, 0)::int AS attempts, + COALESCE(t.dispatch_priority, 100)::int AS priority, + NULLIF(COALESCE(t.error_message, t.skip_reason), '')::text AS error_message, + t.created_at, + t.updated_at, + t.leased_at AS started_at, + COALESCE(t.completed_at, t.callback_received_at) AS completed_at, + t.lease_expires_at, + COALESCE(t.dispatch_after, t.planned_at) AS available_at, + NULL::timestamptz AS last_heartbeat_at + FROM monitoring_collect_tasks t + LEFT JOIN monitoring_collect_requests r ON r.request_id = t.superseded_by_request_id +) +SELECT + source, id, numeric_id, uuid, kind, status, phase, + tenant_id, tenant_name, workspace_id, workspace_name, user_id, user_name, + title, target, queue, worker, attempts, priority, error_message, + created_at, updated_at, started_at, completed_at, lease_expires_at, + available_at, last_heartbeat_at +FROM jobs +WHERE ($%d::text IS NULL OR phase = $%d) + AND ($%d::text IS NULL OR status = $%d) + AND ($%d::text IS NULL OR kind ILIKE '%%' || $%d || '%%') + AND ($%d::text IS NULL OR + id ILIKE '%%' || $%d || '%%' OR + COALESCE(title, '') ILIKE '%%' || $%d || '%%' OR + COALESCE(target, '') ILIKE '%%' || $%d || '%%' OR + COALESCE(error_message, '') ILIKE '%%' || $%d || '%%') + AND ($%d::bigint IS NULL OR tenant_id = $%d) + AND ($%d::bigint IS NULL OR workspace_id = $%d) + AND ($%d::bigint IS NULL OR user_id = $%d) + AND ($%d::timestamptz IS NULL OR created_at >= $%d) + AND ($%d::timestamptz IS NULL OR created_at < $%d) +ORDER BY created_at DESC, id DESC +LIMIT 2000`, + phaseArg, phaseArg, + statusArg, statusArg, + kindArg, kindArg, + keywordArg, keywordArg, keywordArg, keywordArg, keywordArg, + tenantArg, tenantArg, + workspaceArg, workspaceArg, + userArg, userArg, + startArg, startArg, + endArg, endArg, + ), args...) + if err != nil { + if isUndefinedMonitoringRelation(err) { + return nil, nil + } + return nil, err + } + defer rows.Close() + + items := make([]domain.JobSummary, 0) + for rows.Next() { + item, err := scanJobSummary(rows) + if err != nil { + return nil, err + } + items = append(items, *item) + } + return items, rows.Err() +} + +func businessJobUnionSQL() string { + return strings.Join([]string{ + generationJobSelectSQL(), + templateAssistJobSelectSQL(), + kolAssistJobSelectSQL(), + knowledgeParseJobSelectSQL(), + desktopPublishJobSelectSQL(), + desktopTaskJobSelectSQL(), + complianceReviewJobSelectSQL(), + }, "\nUNION ALL\n") +} + +func generationJobSelectSQL() string { + return ` + SELECT + 'generation'::text AS source, + gt.id::text AS id, + gt.id::bigint AS numeric_id, + NULL::text AS uuid, + gt.task_type::text AS kind, + gt.status::text AS status, + CASE + WHEN gt.status = 'queued' THEN 'queued' + WHEN gt.status = 'running' THEN 'running' + WHEN gt.status = 'completed' THEN 'succeeded' + WHEN gt.status = 'failed' THEN 'failed' + WHEN gt.status IN ('cancelled', 'canceled', 'aborted') THEN 'canceled' + ELSE 'unknown' + END AS phase, + gt.tenant_id::bigint AS tenant_id, + t.name::text AS tenant_name, + w.id::bigint AS workspace_id, + w.name::text AS workspace_name, + gt.operator_id::bigint AS user_id, + u.name::text AS user_name, + COALESCE(av.title, 'Article #' || gt.article_id::text, 'Generation task #' || gt.id::text)::text AS title, + CASE WHEN gt.article_id IS NOT NULL THEN 'article #' || gt.article_id::text ELSE NULL END::text AS target, + 'generation.task.run'::text AS queue, + gt.lease_owner::text AS worker, + COALESCE(gt.attempt_count, 0)::int AS attempts, + NULL::int AS priority, + NULLIF(gt.error_message, '')::text AS error_message, + gt.created_at, + gt.updated_at, + gt.started_at, + gt.completed_at, + gt.lease_expires_at, + NULL::timestamptz AS available_at, + gt.last_heartbeat_at + FROM generation_tasks gt + LEFT JOIN tenants t ON t.id = gt.tenant_id + LEFT JOIN users u ON u.id = gt.operator_id + LEFT JOIN articles a ON a.id = gt.article_id + LEFT JOIN article_versions av ON av.id = a.current_version_id + LEFT JOIN LATERAL ( + SELECT wm.workspace_id AS id, ws.name + FROM workspace_memberships wm + JOIN workspaces ws ON ws.id = wm.workspace_id + WHERE wm.tenant_id = gt.tenant_id + AND wm.user_id = gt.operator_id + AND wm.is_primary = TRUE + ORDER BY wm.created_at ASC, wm.id ASC + LIMIT 1 + ) w ON TRUE` +} + +func templateAssistJobSelectSQL() string { + return ` + SELECT + 'template_assist'::text AS source, + tat.id::text AS id, + NULL::bigint AS numeric_id, + tat.id::text AS uuid, + tat.task_type::text AS kind, + tat.status::text AS status, + CASE + WHEN tat.status = 'queued' THEN 'queued' + WHEN tat.status = 'running' THEN 'running' + WHEN tat.status = 'completed' THEN 'succeeded' + WHEN tat.status = 'failed' THEN 'failed' + WHEN tat.status IN ('cancelled', 'canceled', 'aborted') THEN 'canceled' + ELSE 'unknown' + END AS phase, + tat.tenant_id::bigint AS tenant_id, + t.name::text AS tenant_name, + w.id::bigint AS workspace_id, + w.name::text AS workspace_name, + tat.operator_id::bigint AS user_id, + u.name::text AS user_name, + COALESCE(at.template_name, 'Template #' || tat.template_id::text)::text AS title, + ('template #' || tat.template_id::text)::text AS target, + 'template.assist.run'::text AS queue, + NULL::text AS worker, + NULL::int AS attempts, + NULL::int AS priority, + NULLIF(tat.error_message, '')::text AS error_message, + tat.created_at, + tat.updated_at, + tat.started_at, + tat.completed_at, + NULL::timestamptz AS lease_expires_at, + NULL::timestamptz AS available_at, + NULL::timestamptz AS last_heartbeat_at + FROM template_assist_tasks tat + LEFT JOIN tenants t ON t.id = tat.tenant_id + LEFT JOIN users u ON u.id = tat.operator_id + LEFT JOIN article_templates at ON at.id = tat.template_id + LEFT JOIN LATERAL ( + SELECT wm.workspace_id AS id, ws.name + FROM workspace_memberships wm + JOIN workspaces ws ON ws.id = wm.workspace_id + WHERE wm.tenant_id = tat.tenant_id + AND wm.user_id = tat.operator_id + AND wm.is_primary = TRUE + ORDER BY wm.created_at ASC, wm.id ASC + LIMIT 1 + ) w ON TRUE` +} + +func kolAssistJobSelectSQL() string { + return ` + SELECT + 'kol_assist'::text AS source, + kat.id::text AS id, + NULL::bigint AS numeric_id, + kat.id::text AS uuid, + kat.task_type::text AS kind, + kat.status::text AS status, + CASE + WHEN kat.status = 'queued' THEN 'queued' + WHEN kat.status = 'running' THEN 'running' + WHEN kat.status = 'completed' THEN 'succeeded' + WHEN kat.status = 'failed' THEN 'failed' + WHEN kat.status IN ('cancelled', 'canceled', 'aborted') THEN 'canceled' + ELSE 'unknown' + END AS phase, + kat.tenant_id::bigint AS tenant_id, + t.name::text AS tenant_name, + w.id::bigint AS workspace_id, + w.name::text AS workspace_name, + kat.operator_id::bigint AS user_id, + u.name::text AS user_name, + COALESCE(kpr.name, kp.display_name, 'KOL assist #' || kat.id)::text AS title, + COALESCE(kp.display_name, 'profile #' || kat.kol_profile_id::text)::text AS target, + 'kol.assist.run'::text AS queue, + NULL::text AS worker, + NULL::int AS attempts, + NULL::int AS priority, + NULLIF(kat.error_message, '')::text AS error_message, + kat.created_at, + kat.updated_at, + kat.started_at, + kat.completed_at, + NULL::timestamptz AS lease_expires_at, + NULL::timestamptz AS available_at, + NULL::timestamptz AS last_heartbeat_at + FROM kol_assist_tasks kat + LEFT JOIN tenants t ON t.id = kat.tenant_id + LEFT JOIN users u ON u.id = kat.operator_id + LEFT JOIN kol_profiles kp ON kp.id = kat.kol_profile_id + LEFT JOIN kol_prompts kpr ON kpr.id = kat.prompt_id + LEFT JOIN LATERAL ( + SELECT wm.workspace_id AS id, ws.name + FROM workspace_memberships wm + JOIN workspaces ws ON ws.id = wm.workspace_id + WHERE wm.tenant_id = kat.tenant_id + AND wm.user_id = kat.operator_id + AND wm.is_primary = TRUE + ORDER BY wm.created_at ASC, wm.id ASC + LIMIT 1 + ) w ON TRUE` +} + +func knowledgeParseJobSelectSQL() string { + return ` + SELECT + 'knowledge_parse'::text AS source, + kpt.id::text AS id, + kpt.id::bigint AS numeric_id, + NULL::text AS uuid, + ('knowledge:' || kpt.source_type)::text AS kind, + kpt.status::text AS status, + CASE + WHEN kpt.status = 'pending' THEN 'queued' + WHEN kpt.status = 'running' THEN 'running' + WHEN kpt.status = 'completed' THEN 'succeeded' + WHEN kpt.status = 'failed' THEN 'failed' + WHEN kpt.status IN ('deleted', 'cancelled', 'canceled', 'aborted') THEN 'canceled' + ELSE 'unknown' + END AS phase, + kpt.tenant_id::bigint AS tenant_id, + t.name::text AS tenant_name, + NULL::bigint AS workspace_id, + NULL::text AS workspace_name, + NULL::bigint AS user_id, + NULL::text AS user_name, + COALESCE(ki.name, 'Knowledge item #' || kpt.knowledge_item_id::text)::text AS title, + ('knowledge item #' || kpt.knowledge_item_id::text)::text AS target, + 'in_process.knowledge_parse'::text AS queue, + NULL::text AS worker, + NULL::int AS attempts, + NULL::int AS priority, + NULLIF(kpt.error_message, '')::text AS error_message, + kpt.created_at, + kpt.updated_at, + kpt.started_at, + kpt.completed_at, + NULL::timestamptz AS lease_expires_at, + NULL::timestamptz AS available_at, + NULL::timestamptz AS last_heartbeat_at + FROM knowledge_parse_tasks kpt + LEFT JOIN tenants t ON t.id = kpt.tenant_id + LEFT JOIN knowledge_items ki ON ki.id = kpt.knowledge_item_id` +} + +func desktopPublishJobSelectSQL() string { + return ` + SELECT + 'desktop_publish'::text AS source, + j.desktop_id::text AS id, + j.id::bigint AS numeric_id, + j.desktop_id::text AS uuid, + 'publish_job'::text AS kind, + COALESCE(j.status, 'queued')::text AS status, + CASE + WHEN COALESCE(j.status, 'queued') = 'queued' THEN 'queued' + WHEN j.status = 'blocked_by_compliance' THEN 'blocked' + WHEN j.status IN ('completed', 'succeeded') THEN 'succeeded' + WHEN j.status = 'failed' THEN 'failed' + WHEN j.status IN ('cancelled', 'canceled', 'aborted') THEN 'canceled' + ELSE 'unknown' + END AS phase, + j.tenant_id::bigint AS tenant_id, + t.name::text AS tenant_name, + j.workspace_id::bigint AS workspace_id, + w.name::text AS workspace_name, + j.created_by_user_id::bigint AS user_id, + u.name::text AS user_name, + j.title::text AS title, + CASE WHEN j.article_id IS NOT NULL THEN 'article #' || j.article_id::text ELSE NULL END::text AS target, + 'desktop.publish'::text AS queue, + NULL::text AS worker, + NULL::int AS attempts, + NULL::int AS priority, + NULLIF(j.compliance_blocked_reason, '')::text AS error_message, + j.created_at, + j.updated_at, + NULL::timestamptz AS started_at, + NULL::timestamptz AS completed_at, + NULL::timestamptz AS lease_expires_at, + j.scheduled_at AS available_at, + NULL::timestamptz AS last_heartbeat_at + FROM desktop_publish_jobs j + LEFT JOIN tenants t ON t.id = j.tenant_id + LEFT JOIN workspaces w ON w.id = j.workspace_id + LEFT JOIN users u ON u.id = j.created_by_user_id` +} + +func desktopTaskJobSelectSQL() string { + return ` + SELECT + 'desktop_task'::text AS source, + dt.desktop_id::text AS id, + dt.id::bigint AS numeric_id, + dt.desktop_id::text AS uuid, + dt.kind::text AS kind, + dt.status::text AS status, + CASE + WHEN dt.status = 'queued' THEN 'queued' + WHEN dt.status = 'in_progress' THEN 'running' + WHEN dt.status = 'succeeded' THEN 'succeeded' + WHEN dt.status = 'failed' THEN 'failed' + WHEN dt.status = 'aborted' THEN 'canceled' + WHEN dt.status = 'unknown' THEN 'unknown' + ELSE 'unknown' + END AS phase, + dt.tenant_id::bigint AS tenant_id, + t.name::text AS tenant_name, + dt.workspace_id::bigint AS workspace_id, + w.name::text AS workspace_name, + dc.user_id::bigint AS user_id, + u.name::text AS user_name, + COALESCE(j.title, dt.kind || ' task #' || dt.id::text)::text AS title, + COALESCE(mp.name, dt.platform_id, dt.target_client_id::text)::text AS target, + ('desktop.' || COALESCE(dt.lane, 'normal'))::text AS queue, + dt.target_client_id::text AS worker, + COALESCE(dt.attempts, 0)::int AS attempts, + COALESCE(dt.priority, 0)::int AS priority, + NULLIF(dt.error::text, '')::text AS error_message, + dt.created_at, + dt.updated_at, + dt.started_at, + CASE WHEN dt.status IN ('succeeded','failed','aborted') THEN dt.updated_at ELSE NULL END AS completed_at, + dt.lease_expires_at, + dt.enqueued_at AS available_at, + NULL::timestamptz AS last_heartbeat_at + FROM desktop_tasks dt + LEFT JOIN desktop_publish_jobs j ON j.desktop_id = dt.job_id + LEFT JOIN tenants t ON t.id = dt.tenant_id + LEFT JOIN workspaces w ON w.id = dt.workspace_id + LEFT JOIN desktop_clients dc ON dc.id = dt.target_client_id + LEFT JOIN users u ON u.id = dc.user_id + LEFT JOIN media_platforms mp ON mp.platform_id = dt.platform_id` +} + +func complianceReviewJobSelectSQL() string { + return ` + SELECT + 'compliance_review'::text AS source, + crj.id::text AS id, + crj.id::bigint AS numeric_id, + crj.message_id::text AS uuid, + 'compliance_review'::text AS kind, + crj.status::text AS status, + CASE + WHEN crj.status = 'pending' THEN 'queued' + WHEN crj.status = 'running' THEN 'running' + WHEN crj.status = 'done' THEN 'succeeded' + WHEN crj.status = 'failed' THEN 'failed' + ELSE 'unknown' + END AS phase, + crj.tenant_id::bigint AS tenant_id, + t.name::text AS tenant_name, + w.id::bigint AS workspace_id, + w.name::text AS workspace_name, + ccr.checked_by::bigint AS user_id, + u.name::text AS user_name, + COALESCE(av.title, 'Compliance review #' || crj.id::text)::text AS title, + ('check record #' || crj.check_record_id::text)::text AS target, + 'compliance.review.run'::text AS queue, + crj.locked_by::text AS worker, + COALESCE(crj.attempts, 0)::int AS attempts, + NULL::int AS priority, + NULLIF(crj.last_error, '')::text AS error_message, + crj.created_at, + crj.updated_at, + crj.locked_at AS started_at, + CASE WHEN crj.status IN ('done','failed') THEN crj.updated_at ELSE NULL END AS completed_at, + NULL::timestamptz AS lease_expires_at, + crj.available_at, + NULL::timestamptz AS last_heartbeat_at + FROM compliance_review_jobs crj + LEFT JOIN tenants t ON t.id = crj.tenant_id + LEFT JOIN compliance_check_records ccr ON ccr.id = crj.check_record_id + LEFT JOIN users u ON u.id = ccr.checked_by + LEFT JOIN article_versions av ON av.id = crj.article_version_id + LEFT JOIN LATERAL ( + SELECT wm.workspace_id AS id, ws.name + FROM workspace_memberships wm + JOIN workspaces ws ON ws.id = wm.workspace_id + WHERE wm.tenant_id = crj.tenant_id + AND wm.user_id = ccr.checked_by + AND wm.is_primary = TRUE + ORDER BY wm.created_at ASC, wm.id ASC + LIMIT 1 + ) w ON TRUE` +} + +func (r *JobRepository) getGenerationJob(ctx context.Context, id string) (*domain.JobDetail, error) { + jobID, err := parseInt64ID(id) + if err != nil { + return nil, ErrJobNotFound + } + return r.getBusinessJob(ctx, "generation", jobID, ` + SELECT + gt.input_params_json, + NULL::jsonb, + CASE WHEN gt.error_message IS NULL OR gt.error_message = '' THEN NULL ELSE to_jsonb(gt.error_message) END, + jsonb_strip_nulls(jsonb_build_object( + 'article_id', gt.article_id, + 'quota_reservation_id', gt.quota_reservation_id, + 'request_hash', gt.request_hash, + 'task_batch_id', gt.task_batch_id + )) + FROM generation_tasks gt + WHERE gt.id = $1`, jobID) +} + +func (r *JobRepository) getTemplateAssistJob(ctx context.Context, id string) (*domain.JobDetail, error) { + return r.getBusinessJobByTextID(ctx, "template_assist", id, ` + SELECT + tat.request_json, + tat.result_json, + CASE WHEN tat.error_message IS NULL OR tat.error_message = '' THEN NULL ELSE to_jsonb(tat.error_message) END, + jsonb_strip_nulls(jsonb_build_object( + 'template_id', tat.template_id, + 'template_key', at.template_key, + 'template_name', at.template_name, + 'analyze_prompt', COALESCE( + at.card_config_json #>> '{wizard,structure,analyze_prompt_template}', + at.card_config_json #>> '{wizard,analyze_prompt_template}' + ), + 'title_prompt', COALESCE( + at.card_config_json #>> '{wizard,structure,title_prompt_template}', + at.card_config_json #>> '{wizard,title_prompt_template}' + ), + 'outline_prompt', COALESCE( + at.card_config_json #>> '{wizard,structure,outline_prompt_template}', + at.card_config_json #>> '{wizard,outline_prompt_template}' + ) + )) + FROM template_assist_tasks tat + LEFT JOIN article_templates at ON at.id = tat.template_id + WHERE tat.id = $1`) +} + +func (r *JobRepository) getKolAssistJob(ctx context.Context, id string) (*domain.JobDetail, error) { + return r.getBusinessJobByTextID(ctx, "kol_assist", id, ` + SELECT + kat.request_json, + kat.result_json, + CASE WHEN kat.error_message IS NULL OR kat.error_message = '' THEN NULL ELSE to_jsonb(kat.error_message) END, + jsonb_strip_nulls(jsonb_build_object( + 'kol_profile_id', kat.kol_profile_id, + 'prompt_id', kat.prompt_id + )) + FROM kol_assist_tasks kat + WHERE kat.id = $1`) +} + +func (r *JobRepository) getKnowledgeParseJob(ctx context.Context, id string) (*domain.JobDetail, error) { + jobID, err := parseInt64ID(id) + if err != nil { + return nil, ErrJobNotFound + } + return r.getBusinessJob(ctx, "knowledge_parse", jobID, ` + SELECT + jsonb_strip_nulls(jsonb_build_object( + 'knowledge_item_id', kpt.knowledge_item_id, + 'source_type', kpt.source_type, + 'source_uri', ki.source_uri, + 'storage_key', ki.storage_key + )), + NULL::jsonb, + CASE WHEN kpt.error_message IS NULL OR kpt.error_message = '' THEN NULL ELSE to_jsonb(kpt.error_message) END, + jsonb_strip_nulls(jsonb_build_object( + 'group_id', ki.group_id, + 'item_status', ki.status, + 'size_bytes', ki.size_bytes, + 'item_version', ki.item_version + )) + FROM knowledge_parse_tasks kpt + LEFT JOIN knowledge_items ki ON ki.id = kpt.knowledge_item_id + WHERE kpt.id = $1`, jobID) +} + +func (r *JobRepository) getDesktopPublishJob(ctx context.Context, id string) (*domain.JobDetail, error) { + return r.getBusinessJobByTextID(ctx, "desktop_publish", id, ` + SELECT + j.content_ref, + NULL::jsonb, + CASE WHEN j.compliance_blocked_reason IS NULL OR j.compliance_blocked_reason = '' THEN NULL ELSE to_jsonb(j.compliance_blocked_reason) END, + jsonb_strip_nulls(jsonb_build_object( + 'article_id', j.article_id, + 'article_version_id', j.article_version_id, + 'compliance_blocked_record_id', j.compliance_blocked_record_id, + 'compliance_blocked_at', j.compliance_blocked_at, + 'scheduled_at', j.scheduled_at + )) + FROM desktop_publish_jobs j + WHERE j.desktop_id::text = $1`) +} + +func (r *JobRepository) getDesktopTaskJob(ctx context.Context, id string) (*domain.JobDetail, error) { + return r.getBusinessJobByTextID(ctx, "desktop_task", id, ` + SELECT + dt.payload, + dt.result, + dt.error, + jsonb_strip_nulls(jsonb_build_object( + 'job_id', dt.job_id, + 'target_account_id', dt.target_account_id, + 'target_client_id', dt.target_client_id, + 'platform_id', dt.platform_id, + 'dedup_key', dt.dedup_key, + 'active_attempt_id', dt.active_attempt_id, + 'lane', dt.lane, + 'lane_weight', dt.lane_weight, + 'source', dt.source, + 'scheduler_group_key', dt.scheduler_group_key, + 'monitor_task_id', dt.monitor_task_id, + 'supersedes_task_id', dt.supersedes_task_id, + 'control_flags', dt.control_flags, + 'interrupt_generation', dt.interrupt_generation, + 'interrupted_at', dt.interrupted_at, + 'interrupt_reason', dt.interrupt_reason + )) + FROM desktop_tasks dt + WHERE dt.desktop_id::text = $1`) +} + +func (r *JobRepository) getComplianceReviewJob(ctx context.Context, id string) (*domain.JobDetail, error) { + jobID, err := parseInt64ID(id) + if err != nil { + return nil, ErrJobNotFound + } + return r.getBusinessJob(ctx, "compliance_review", jobID, ` + SELECT + jsonb_strip_nulls(jsonb_build_object( + 'message_id', crj.message_id, + 'article_id', crj.article_id, + 'article_version_id', crj.article_version_id, + 'check_record_id', crj.check_record_id + )), + NULL::jsonb, + CASE WHEN crj.last_error IS NULL OR crj.last_error = '' THEN NULL ELSE to_jsonb(crj.last_error) END, + jsonb_strip_nulls(jsonb_build_object( + 'available_at', crj.available_at, + 'locked_at', crj.locked_at, + 'locked_by', crj.locked_by + )) + FROM compliance_review_jobs crj + WHERE crj.id = $1`, jobID) +} + +func (r *JobRepository) getMonitoringCollectJob(ctx context.Context, id string) (*domain.JobDetail, error) { + if r == nil || r.monitoringPool == nil { + return nil, ErrJobNotFound + } + jobID, err := parseInt64ID(id) + if err != nil { + return nil, ErrJobNotFound + } + summary, err := r.getMonitoringSummary(ctx, jobID) + if err != nil { + return nil, err + } + + var payloadRaw, metaRaw []byte + err = r.monitoringPool.QueryRow(ctx, ` + SELECT + request_payload_json, + jsonb_strip_nulls(jsonb_build_object( + 'brand_id', brand_id, + 'question_id', question_id, + 'question_hash', encode(question_hash, 'hex'), + 'ai_platform_id', ai_platform_id, + 'collector_type', collector_type, + 'trigger_source', trigger_source, + 'run_mode', run_mode, + 'business_date', business_date, + 'planned_at', planned_at, + 'lease_token_hash', lease_token_hash, + 'leased_to_executor', leased_to_executor, + 'callback_received_at', callback_received_at, + 'skip_reason', skip_reason, + 'dispatch_priority', dispatch_priority, + 'dispatch_lane', dispatch_lane, + 'target_client_id', target_client_id, + 'dispatch_after', dispatch_after, + 'interrupt_generation', interrupt_generation, + 'superseded_by_request_id', superseded_by_request_id, + 'last_dispatched_at', last_dispatched_at, + 'dispatch_attempts', dispatch_attempts, + 'ingest_shard_key', ingest_shard_key, + 'execution_owner', execution_owner + )) + FROM monitoring_collect_tasks + WHERE id = $1 + `, jobID).Scan(&payloadRaw, &metaRaw) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrJobNotFound + } + if isUndefinedMonitoringRelation(err) { + return nil, ErrJobNotFound + } + return nil, err + } + + return &domain.JobDetail{ + JobSummary: *summary, + Payload: decodeJSONValue(payloadRaw), + Metadata: decodeJSONMap(metaRaw), + }, nil +} + +func (r *JobRepository) getBusinessJob(ctx context.Context, source string, numericID int64, detailSQL string, args ...any) (*domain.JobDetail, error) { + summary, err := r.getBusinessSummary(ctx, source, strconv.FormatInt(numericID, 10)) + if err != nil { + return nil, err + } + return r.attachBusinessDetail(ctx, summary, detailSQL, args...) +} + +func (r *JobRepository) getBusinessJobByTextID(ctx context.Context, source, id, detailSQL string) (*domain.JobDetail, error) { + summary, err := r.getBusinessSummary(ctx, source, id) + if err != nil { + return nil, err + } + return r.attachBusinessDetail(ctx, summary, detailSQL, id) +} + +func (r *JobRepository) getBusinessSummary(ctx context.Context, source, id string) (*domain.JobSummary, error) { + items, err := r.listBusinessJobs(ctx, domain.JobFilter{ + Source: source, + Keyword: id, + Limit: 2000, + }) + if err != nil { + return nil, err + } + for i := range items { + if items[i].Source == source && items[i].ID == id { + return &items[i], nil + } + } + return nil, ErrJobNotFound +} + +func (r *JobRepository) getMonitoringSummary(ctx context.Context, id int64) (*domain.JobSummary, error) { + items, err := r.listMonitoringJobs(ctx, domain.JobFilter{ + Source: "monitoring_collect", + Keyword: strconv.FormatInt(id, 10), + Limit: 2000, + }) + if err != nil { + return nil, err + } + for i := range items { + if items[i].ID == strconv.FormatInt(id, 10) { + return &items[i], nil + } + } + return nil, ErrJobNotFound +} + +func (r *JobRepository) attachBusinessDetail(ctx context.Context, summary *domain.JobSummary, query string, args ...any) (*domain.JobDetail, error) { + if summary == nil { + return nil, ErrJobNotFound + } + var payloadRaw, resultRaw, errorRaw, metaRaw []byte + err := r.pool.QueryRow(ctx, query, args...).Scan(&payloadRaw, &resultRaw, &errorRaw, &metaRaw) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrJobNotFound + } + return nil, err + } + return &domain.JobDetail{ + JobSummary: *summary, + Payload: decodeJSONValue(payloadRaw), + Result: decodeJSONValue(resultRaw), + Error: decodeJSONValue(errorRaw), + Metadata: decodeJSONMap(metaRaw), + }, nil +} + +func (r *JobRepository) retryGenerationJob(ctx context.Context, id string) (*domain.JobDetail, error) { + jobID, err := parseInt64ID(id) + if err != nil { + return nil, ErrJobNotFound + } + cmd, err := r.pool.Exec(ctx, ` + UPDATE generation_tasks + SET status = 'queued', + error_message = NULL, + started_at = NULL, + completed_at = NULL, + lease_token = NULL, + lease_owner = NULL, + lease_expires_at = NULL, + updated_at = NOW() + WHERE id = $1 + AND status = 'failed' + `, jobID) + if err != nil { + return nil, err + } + if cmd.RowsAffected() == 0 { + return nil, ErrJobNotFound + } + return r.getGenerationJob(ctx, id) +} + +func (r *JobRepository) retryTemplateAssistJob(ctx context.Context, id string) (*domain.JobDetail, error) { + cmd, err := r.pool.Exec(ctx, ` + UPDATE template_assist_tasks + SET status = 'queued', + result_json = NULL, + error_message = NULL, + started_at = NULL, + completed_at = NULL, + updated_at = NOW() + WHERE id = $1 + AND status = 'failed' + `, strings.TrimSpace(id)) + if err != nil { + return nil, err + } + if cmd.RowsAffected() == 0 { + return nil, ErrJobNotFound + } + return r.getTemplateAssistJob(ctx, id) +} + +func (r *JobRepository) retryKolAssistJob(ctx context.Context, id string) (*domain.JobDetail, error) { + cmd, err := r.pool.Exec(ctx, ` + UPDATE kol_assist_tasks + SET status = 'queued', + result_json = NULL, + error_message = NULL, + started_at = NULL, + completed_at = NULL, + updated_at = NOW() + WHERE id = $1 + AND status = 'failed' + `, strings.TrimSpace(id)) + if err != nil { + return nil, err + } + if cmd.RowsAffected() == 0 { + return nil, ErrJobNotFound + } + return r.getKolAssistJob(ctx, id) +} + +func (r *JobRepository) retryKnowledgeParseJob(ctx context.Context, id string) (*domain.JobDetail, error) { + jobID, err := parseInt64ID(id) + if err != nil { + return nil, ErrJobNotFound + } + cmd, err := r.pool.Exec(ctx, ` + WITH target AS ( + SELECT kpt.id, kpt.tenant_id, kpt.knowledge_item_id + FROM knowledge_parse_tasks kpt + WHERE kpt.id = $1 + AND kpt.status = 'failed' + ), + task_update AS ( + UPDATE knowledge_parse_tasks kpt + SET status = 'pending', + error_message = NULL, + started_at = NULL, + completed_at = NULL, + updated_at = NOW() + FROM target + WHERE kpt.id = target.id + RETURNING target.tenant_id, target.knowledge_item_id + ) + UPDATE knowledge_items ki + SET status = 'pending', + error_message = NULL, + updated_at = NOW() + FROM task_update + WHERE ki.id = task_update.knowledge_item_id + AND ki.tenant_id = task_update.tenant_id + `, jobID) + if err != nil { + return nil, err + } + if cmd.RowsAffected() == 0 { + return nil, ErrJobNotFound + } + return r.getKnowledgeParseJob(ctx, id) +} + +func (r *JobRepository) retryDesktopTaskJob(ctx context.Context, id string) (*domain.JobDetail, error) { + errorJSON, _ := json.Marshal(map[string]any{ + "message": "retried from ops job console", + "retried_at": time.Now().UTC().Format(time.RFC3339), + }) + cmd, err := r.pool.Exec(ctx, ` + UPDATE desktop_tasks + SET status = 'queued', + error = $2, + result = NULL, + active_attempt_id = NULL, + lease_token_hash = NULL, + lease_expires_at = NULL, + attempts = attempts + 1, + started_at = NULL, + interrupted_at = NULL, + interrupt_reason = NULL, + enqueued_at = NOW(), + updated_at = NOW() + WHERE desktop_id::text = $1 + AND status IN ('failed', 'unknown', 'aborted') + `, strings.TrimSpace(id), errorJSON) + if err != nil { + return nil, err + } + if cmd.RowsAffected() == 0 { + return nil, ErrJobNotFound + } + return r.getDesktopTaskJob(ctx, id) +} + +func (r *JobRepository) retryComplianceReviewJob(ctx context.Context, id string) (*domain.JobDetail, error) { + jobID, err := parseInt64ID(id) + if err != nil { + return nil, ErrJobNotFound + } + cmd, err := r.pool.Exec(ctx, ` + UPDATE compliance_review_jobs + SET status = 'pending', + available_at = NOW(), + locked_at = NULL, + locked_by = NULL, + last_error = NULL, + updated_at = NOW() + WHERE id = $1 + AND status = 'failed' + `, jobID) + if err != nil { + return nil, err + } + if cmd.RowsAffected() == 0 { + return nil, ErrJobNotFound + } + return r.getComplianceReviewJob(ctx, id) +} + +func (r *JobRepository) retryMonitoringCollectJob(ctx context.Context, id string) (*domain.JobDetail, error) { + if r == nil || r.monitoringPool == nil { + return nil, ErrJobNotFound + } + jobID, err := parseInt64ID(id) + if err != nil { + return nil, ErrJobNotFound + } + cmd, err := r.monitoringPool.Exec(ctx, ` + UPDATE monitoring_collect_tasks + SET status = 'pending', + lease_token_hash = NULL, + leased_to_executor = NULL, + leased_at = NULL, + lease_expires_at = NULL, + callback_received_at = NULL, + completed_at = NULL, + skip_reason = NULL, + error_message = NULL, + dispatch_after = NOW(), + updated_at = NOW() + WHERE id = $1 + AND status IN ('failed', 'expired', 'skipped') + `, jobID) + if err != nil { + if isUndefinedMonitoringRelation(err) { + return nil, ErrJobNotFound + } + return nil, err + } + if cmd.RowsAffected() == 0 { + return nil, ErrJobNotFound + } + return r.getMonitoringCollectJob(ctx, id) +} + +func (r *JobRepository) cancelGenerationJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) { + jobID, err := parseInt64ID(id) + if err != nil { + return nil, ErrJobNotFound + } + cmd, err := r.pool.Exec(ctx, ` + UPDATE generation_tasks + SET status = 'failed', + error_message = $2, + completed_at = NOW(), + lease_token = NULL, + lease_owner = NULL, + lease_expires_at = NULL, + updated_at = NOW() + WHERE id = $1 + AND status IN ('queued', 'running') + `, jobID, reason) + if err != nil { + return nil, err + } + if cmd.RowsAffected() == 0 { + return nil, ErrJobNotFound + } + return r.getGenerationJob(ctx, id) +} + +func (r *JobRepository) cancelTemplateAssistJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) { + cmd, err := r.pool.Exec(ctx, ` + UPDATE template_assist_tasks + SET status = 'failed', + error_message = $2, + completed_at = NOW(), + updated_at = NOW() + WHERE id = $1 + AND status IN ('queued', 'running') + `, strings.TrimSpace(id), reason) + if err != nil { + return nil, err + } + if cmd.RowsAffected() == 0 { + return nil, ErrJobNotFound + } + return r.getTemplateAssistJob(ctx, id) +} + +func (r *JobRepository) cancelKolAssistJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) { + cmd, err := r.pool.Exec(ctx, ` + UPDATE kol_assist_tasks + SET status = 'failed', + error_message = $2, + completed_at = NOW(), + updated_at = NOW() + WHERE id = $1 + AND status IN ('queued', 'running') + `, strings.TrimSpace(id), reason) + if err != nil { + return nil, err + } + if cmd.RowsAffected() == 0 { + return nil, ErrJobNotFound + } + return r.getKolAssistJob(ctx, id) +} + +func (r *JobRepository) cancelKnowledgeParseJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) { + jobID, err := parseInt64ID(id) + if err != nil { + return nil, ErrJobNotFound + } + cmd, err := r.pool.Exec(ctx, ` + WITH target AS ( + SELECT kpt.id, kpt.tenant_id, kpt.knowledge_item_id + FROM knowledge_parse_tasks kpt + WHERE kpt.id = $1 + AND kpt.status IN ('pending', 'running') + ), + task_update AS ( + UPDATE knowledge_parse_tasks kpt + SET status = 'deleted', + error_message = $2, + completed_at = NOW(), + updated_at = NOW() + FROM target + WHERE kpt.id = target.id + RETURNING target.tenant_id, target.knowledge_item_id + ) + UPDATE knowledge_items ki + SET status = 'failed', + error_message = $2, + updated_at = NOW() + FROM task_update + WHERE ki.id = task_update.knowledge_item_id + AND ki.tenant_id = task_update.tenant_id + `, jobID, reason) + if err != nil { + return nil, err + } + if cmd.RowsAffected() == 0 { + return nil, ErrJobNotFound + } + return r.getKnowledgeParseJob(ctx, id) +} + +func (r *JobRepository) cancelDesktopPublishJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) { + tx, err := r.pool.Begin(ctx) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + + cmd, err := tx.Exec(ctx, ` + UPDATE desktop_publish_jobs + SET status = 'aborted', + compliance_blocked_reason = $2, + updated_at = NOW() + WHERE desktop_id::text = $1 + AND status IN ('queued', 'blocked_by_compliance') + `, strings.TrimSpace(id), reason) + if err != nil { + return nil, err + } + if cmd.RowsAffected() == 0 { + return nil, ErrJobNotFound + } + + errorJSON, _ := json.Marshal(map[string]any{"message": reason}) + if _, err := tx.Exec(ctx, ` + UPDATE desktop_tasks + SET status = 'aborted', + error = $2, + active_attempt_id = NULL, + lease_token_hash = NULL, + lease_expires_at = NULL, + interrupted_at = COALESCE(interrupted_at, NOW()), + interrupt_reason = 'ops_cancel', + updated_at = NOW() + WHERE job_id::text = $1 + AND status IN ('queued', 'in_progress', 'unknown') + `, strings.TrimSpace(id), errorJSON); err != nil { + return nil, err + } + + if err := tx.Commit(ctx); err != nil { + return nil, err + } + return r.getDesktopPublishJob(ctx, id) +} + +func (r *JobRepository) cancelDesktopTaskJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) { + errorJSON, _ := json.Marshal(map[string]any{"message": reason}) + cmd, err := r.pool.Exec(ctx, ` + UPDATE desktop_tasks + SET status = 'aborted', + error = $2, + active_attempt_id = NULL, + lease_token_hash = NULL, + lease_expires_at = NULL, + interrupted_at = COALESCE(interrupted_at, NOW()), + interrupt_reason = 'ops_cancel', + updated_at = NOW() + WHERE desktop_id::text = $1 + AND status IN ('queued', 'in_progress', 'unknown') + `, strings.TrimSpace(id), errorJSON) + if err != nil { + return nil, err + } + if cmd.RowsAffected() == 0 { + return nil, ErrJobNotFound + } + return r.getDesktopTaskJob(ctx, id) +} + +func (r *JobRepository) cancelComplianceReviewJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) { + jobID, err := parseInt64ID(id) + if err != nil { + return nil, ErrJobNotFound + } + cmd, err := r.pool.Exec(ctx, ` + UPDATE compliance_review_jobs + SET status = 'failed', + locked_at = NULL, + locked_by = NULL, + last_error = $2, + updated_at = NOW() + WHERE id = $1 + AND status IN ('pending', 'running') + `, jobID, reason) + if err != nil { + return nil, err + } + if cmd.RowsAffected() == 0 { + return nil, ErrJobNotFound + } + return r.getComplianceReviewJob(ctx, id) +} + +func (r *JobRepository) cancelMonitoringCollectJob(ctx context.Context, id, reason string) (*domain.JobDetail, error) { + if r == nil || r.monitoringPool == nil { + return nil, ErrJobNotFound + } + jobID, err := parseInt64ID(id) + if err != nil { + return nil, ErrJobNotFound + } + cmd, err := r.monitoringPool.Exec(ctx, ` + UPDATE monitoring_collect_tasks + SET status = 'skipped', + skip_reason = 'ops_cancel', + error_message = $2, + lease_token_hash = NULL, + leased_to_executor = NULL, + leased_at = NULL, + lease_expires_at = NULL, + completed_at = NOW(), + updated_at = NOW() + WHERE id = $1 + AND status IN ('pending', 'leased', 'received', 'expired') + `, jobID, reason) + if err != nil { + if isUndefinedMonitoringRelation(err) { + return nil, ErrJobNotFound + } + return nil, err + } + if cmd.RowsAffected() == 0 { + return nil, ErrJobNotFound + } + return r.getMonitoringCollectJob(ctx, id) +} + +type jobScanner interface { + Scan(dest ...any) error +} + +func scanJobSummary(row jobScanner) (*domain.JobSummary, error) { + var ( + item domain.JobSummary + numericID pgtype.Int8 + uuidValue pgtype.Text + tenantID pgtype.Int8 + tenantName pgtype.Text + workspaceID pgtype.Int8 + workspaceName pgtype.Text + userID pgtype.Int8 + userName pgtype.Text + title pgtype.Text + target pgtype.Text + queue pgtype.Text + worker pgtype.Text + attempts pgtype.Int4 + priority pgtype.Int4 + errorMessage pgtype.Text + startedAt pgtype.Timestamptz + completedAt pgtype.Timestamptz + leaseExpires pgtype.Timestamptz + availableAt pgtype.Timestamptz + heartbeatAt pgtype.Timestamptz + ) + + if err := row.Scan( + &item.Source, + &item.ID, + &numericID, + &uuidValue, + &item.Kind, + &item.Status, + &item.Phase, + &tenantID, + &tenantName, + &workspaceID, + &workspaceName, + &userID, + &userName, + &title, + &target, + &queue, + &worker, + &attempts, + &priority, + &errorMessage, + &item.CreatedAt, + &item.UpdatedAt, + &startedAt, + &completedAt, + &leaseExpires, + &availableAt, + &heartbeatAt, + ); err != nil { + return nil, err + } + + item.UID = item.Source + ":" + item.ID + item.NumericID = int64PtrFromPg(numericID) + item.UUID = stringPtrFromPg(uuidValue) + item.TenantID = int64PtrFromPg(tenantID) + item.TenantName = stringPtrFromPg(tenantName) + item.WorkspaceID = int64PtrFromPg(workspaceID) + item.WorkspaceName = stringPtrFromPg(workspaceName) + item.UserID = int64PtrFromPg(userID) + item.UserName = stringPtrFromPg(userName) + item.Title = stringPtrFromPg(title) + item.Target = stringPtrFromPg(target) + item.Queue = stringPtrFromPg(queue) + item.Worker = stringPtrFromPg(worker) + item.Attempts = intPtrFromPg(attempts) + item.Priority = intPtrFromPg(priority) + item.ErrorMessage = stringPtrFromPg(errorMessage) + item.StartedAt = timePtrFromPg(startedAt) + item.CompletedAt = timePtrFromPg(completedAt) + item.LeaseExpiresAt = timePtrFromPg(leaseExpires) + item.AvailableAt = timePtrFromPg(availableAt) + item.LastHeartbeatAt = timePtrFromPg(heartbeatAt) + item.DurationSeconds = durationSeconds(item.StartedAt, item.CompletedAt, item.CreatedAt) + item.IsStuck = isJobStuck(item) + item.CanRetry = canRetryJob(item) + item.CanCancel = canCancelJob(item) + return &item, nil +} + +func sortJobs(items []domain.JobSummary) { + for i := 1; i < len(items); i++ { + current := items[i] + j := i - 1 + for j >= 0 && jobLess(current, items[j]) { + items[j+1] = items[j] + j-- + } + items[j+1] = current + } +} + +func jobLess(a, b domain.JobSummary) bool { + if !a.CreatedAt.Equal(b.CreatedAt) { + return a.CreatedAt.After(b.CreatedAt) + } + if a.Source != b.Source { + return a.Source < b.Source + } + return a.ID > b.ID +} + +func buildJobCounts(items []domain.JobSummary) domain.JobCounts { + counts := domain.JobCounts{ + Total: int64(len(items)), + ByPhase: map[string]int64{}, + BySource: map[string]int64{}, + } + for _, item := range items { + counts.ByPhase[item.Phase]++ + counts.BySource[item.Source]++ + } + return counts +} + +func durationSeconds(startedAt, completedAt *time.Time, createdAt time.Time) *int64 { + start := createdAt + if startedAt != nil { + start = *startedAt + } + end := time.Now().UTC() + if completedAt != nil { + end = *completedAt + } + seconds := int64(end.Sub(start).Seconds()) + if seconds < 0 { + seconds = 0 + } + return &seconds +} + +func isJobStuck(item domain.JobSummary) bool { + now := time.Now().UTC() + if item.LeaseExpiresAt != nil && item.LeaseExpiresAt.Before(now) && (item.Phase == domain.JobPhaseRunning || item.Phase == domain.JobPhaseQueued) { + return true + } + if item.Phase == domain.JobPhaseRunning && item.UpdatedAt.Before(now.Add(-30*time.Minute)) { + return true + } + if item.Phase == domain.JobPhaseQueued && item.UpdatedAt.Before(now.Add(-2*time.Hour)) { + return true + } + return false +} + +func canRetryJob(item domain.JobSummary) bool { + switch item.Source { + case "desktop_publish": + return false + case "knowledge_parse": + return false + case "monitoring_collect": + return item.Phase == domain.JobPhaseFailed || item.Phase == domain.JobPhaseCanceled + default: + return item.Phase == domain.JobPhaseFailed || item.Phase == domain.JobPhaseCanceled || item.Phase == domain.JobPhaseUnknown + } +} + +func canCancelJob(item domain.JobSummary) bool { + return item.Phase == domain.JobPhaseQueued || item.Phase == domain.JobPhaseRunning || item.Phase == domain.JobPhaseBlocked || item.Phase == domain.JobPhaseUnknown +} + +func normalizeJobSource(source string) string { + source = strings.TrimSpace(source) + if source == "desktop_publish_job" { + return "desktop_publish" + } + if source == "publish_job" { + return "desktop_publish" + } + return source +} + +func nullableTextValue(value string) *string { + value = strings.TrimSpace(value) + if value == "" { + return nil + } + return &value +} + +func nullableKeyword(value string) *string { + value = strings.TrimSpace(value) + if value == "" { + return nil + } + return &value +} + +func parseInt64ID(id string) (int64, error) { + parsed, err := strconv.ParseInt(strings.TrimSpace(id), 10, 64) + if err != nil || parsed <= 0 { + return 0, fmt.Errorf("invalid id") + } + return parsed, nil +} + +func int64PtrFromPg(value pgtype.Int8) *int64 { + if !value.Valid { + return nil + } + v := value.Int64 + return &v +} + +func intPtrFromPg(value pgtype.Int4) *int { + if !value.Valid { + return nil + } + v := int(value.Int32) + return &v +} + +func stringPtrFromPg(value pgtype.Text) *string { + if !value.Valid { + return nil + } + v := strings.TrimSpace(value.String) + if v == "" { + return nil + } + return &v +} + +func timePtrFromPg(value pgtype.Timestamptz) *time.Time { + if !value.Valid { + return nil + } + v := value.Time + return &v +} + +func decodeJSONValue(raw []byte) any { + if len(raw) == 0 { + return nil + } + var value any + if err := json.Unmarshal(raw, &value); err != nil { + return string(raw) + } + return redactValue(value) +} + +func decodeJSONMap(raw []byte) map[string]any { + value := decodeJSONValue(raw) + if value == nil { + return nil + } + if m, ok := value.(map[string]any); ok { + return m + } + return map[string]any{"value": value} +} + +func redactValue(value any) any { + switch typed := value.(type) { + case map[string]any: + out := make(map[string]any, len(typed)) + for key, nested := range typed { + if isSensitiveKey(key) { + out[key] = "[REDACTED]" + continue + } + out[key] = redactValue(nested) + } + return out + case []any: + out := make([]any, len(typed)) + for i := range typed { + out[i] = redactValue(typed[i]) + } + return out + default: + return value + } +} + +func isSensitiveKey(key string) bool { + normalized := strings.ToLower(strings.TrimSpace(key)) + for _, marker := range []string{ + "password", + "passwd", + "secret", + "token", + "cookie", + "authorization", + "credential", + "session", + "api_key", + "apikey", + "access_key", + "refresh", + "lease_token", + } { + if strings.Contains(normalized, marker) { + return true + } + } + return false +} + +func isUndefinedMonitoringRelation(err error) bool { + var pgErr *pgconn.PgError + return errors.As(err, &pgErr) && (pgErr.Code == "42P01" || pgErr.Code == "42703") +} diff --git a/server/internal/ops/transport/job_handler.go b/server/internal/ops/transport/job_handler.go new file mode 100644 index 0000000..2f0bbf7 --- /dev/null +++ b/server/internal/ops/transport/job_handler.go @@ -0,0 +1,120 @@ +package transport + +import ( + "errors" + "io" + "strconv" + "strings" + "time" + + "github.com/gin-gonic/gin" + + "github.com/geo-platform/tenant-api/internal/ops/app" + "github.com/geo-platform/tenant-api/internal/ops/domain" + "github.com/geo-platform/tenant-api/internal/shared/response" +) + +func listJobsHandler(svc *app.JobService) gin.HandlerFunc { + return func(c *gin.Context) { + page, _ := strconv.Atoi(c.DefaultQuery("page", "1")) + size, _ := strconv.Atoi(c.DefaultQuery("size", "50")) + + filter := domain.JobFilter{ + Source: strings.TrimSpace(c.Query("source")), + Phase: strings.TrimSpace(c.Query("phase")), + Status: strings.TrimSpace(c.Query("status")), + Kind: strings.TrimSpace(c.Query("kind")), + Keyword: strings.TrimSpace(c.Query("keyword")), + } + + if v := strings.TrimSpace(c.Query("tenant_id")); v != "" { + id, err := strconv.ParseInt(v, 10, 64) + if err != nil || id <= 0 { + response.Error(c, response.ErrBadRequest(40070, "invalid_tenant_id", "tenant_id 不合法")) + return + } + filter.TenantID = &id + } + if v := strings.TrimSpace(c.Query("workspace_id")); v != "" { + id, err := strconv.ParseInt(v, 10, 64) + if err != nil || id <= 0 { + response.Error(c, response.ErrBadRequest(40071, "invalid_workspace_id", "workspace_id 不合法")) + return + } + filter.WorkspaceID = &id + } + if v := strings.TrimSpace(c.Query("user_id")); v != "" { + id, err := strconv.ParseInt(v, 10, 64) + if err != nil || id <= 0 { + response.Error(c, response.ErrBadRequest(40072, "invalid_user_id", "user_id 不合法")) + return + } + filter.UserID = &id + } + if v := strings.TrimSpace(c.Query("start_at")); v != "" { + t, err := time.Parse(time.RFC3339, v) + if err != nil { + response.Error(c, response.ErrBadRequest(40073, "invalid_start_at", "start_at 必须是 RFC3339")) + return + } + filter.StartAt = &t + } + if v := strings.TrimSpace(c.Query("end_at")); v != "" { + t, err := time.Parse(time.RFC3339, v) + if err != nil { + response.Error(c, response.ErrBadRequest(40074, "invalid_end_at", "end_at 必须是 RFC3339")) + return + } + filter.EndAt = &t + } + + result, err := svc.List(c.Request.Context(), filter, page, size) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, result) + } +} + +func getJobHandler(svc *app.JobService) gin.HandlerFunc { + return func(c *gin.Context) { + item, err := svc.Get(c.Request.Context(), c.Param("source"), c.Param("id")) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, item) + } +} + +func retryJobHandler(svc *app.JobService) gin.HandlerFunc { + return func(c *gin.Context) { + item, err := svc.Retry(c.Request.Context(), actorFromGin(c), c.Param("source"), c.Param("id")) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, item) + } +} + +func cancelJobHandler(svc *app.JobService) gin.HandlerFunc { + return func(c *gin.Context) { + var req struct { + Reason string `json:"reason"` + } + if c.Request.Body != nil { + if err := c.ShouldBindJSON(&req); err != nil && !errors.Is(err, io.EOF) { + response.Error(c, response.ErrBadRequest(40075, "invalid_payload", err.Error())) + return + } + } + item, err := svc.Cancel(c.Request.Context(), actorFromGin(c), c.Param("source"), c.Param("id"), req.Reason) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, item) + } +} diff --git a/server/internal/ops/transport/router.go b/server/internal/ops/transport/router.go index ef4a703..141fac2 100644 --- a/server/internal/ops/transport/router.go +++ b/server/internal/ops/transport/router.go @@ -19,6 +19,7 @@ type Deps struct { Auth *app.AuthService Accounts *app.AccountService AdminUsers *app.AdminUserService + Jobs *app.JobService KolSubs *app.KolSubscriptionService Audits *app.AuditService SiteDomains *app.SiteDomainMappingService @@ -92,6 +93,11 @@ func RegisterRoutes(d Deps) { authed.GET("/audits", listAuditsHandler(d.Audits)) + authed.GET("/jobs", listJobsHandler(d.Jobs)) + authed.GET("/jobs/:source/:id", getJobHandler(d.Jobs)) + authed.POST("/jobs/:source/:id/retry", retryJobHandler(d.Jobs)) + authed.POST("/jobs/:source/:id/cancel", cancelJobHandler(d.Jobs)) + authed.GET("/site-domain-mappings", listSiteDomainMappingsHandler(d.SiteDomains)) authed.POST("/site-domain-mappings", createSiteDomainMappingHandler(d.SiteDomains)) authed.PATCH("/site-domain-mappings/:id", updateSiteDomainMappingHandler(d.SiteDomains)) diff --git a/server/migrations/20260421100020_create_desktop_tasks.up.sql b/server/migrations/20260421100020_create_desktop_tasks.up.sql index 7b494a6..6d9d8aa 100644 --- a/server/migrations/20260421100020_create_desktop_tasks.up.sql +++ b/server/migrations/20260421100020_create_desktop_tasks.up.sql @@ -11,6 +11,12 @@ CREATE TABLE IF NOT EXISTS desktop_publish_jobs ( scheduled_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + article_id BIGINT REFERENCES articles(id), + article_version_id BIGINT REFERENCES article_versions(id), + status VARCHAR(32) NOT NULL DEFAULT 'queued', + compliance_blocked_record_id BIGINT, + compliance_blocked_at TIMESTAMPTZ, + compliance_blocked_reason TEXT, UNIQUE (desktop_id) ); diff --git a/server/migrations/20260505102000_create_compliance_tables.down.sql b/server/migrations/20260505102000_create_compliance_tables.down.sql index 0c10757..771284f 100644 --- a/server/migrations/20260505102000_create_compliance_tables.down.sql +++ b/server/migrations/20260505102000_create_compliance_tables.down.sql @@ -1,12 +1,7 @@ DROP INDEX IF EXISTS idx_desktop_publish_jobs_status_scheduled; ALTER TABLE desktop_publish_jobs - DROP COLUMN IF EXISTS compliance_blocked_reason, - DROP COLUMN IF EXISTS compliance_blocked_at, - DROP COLUMN IF EXISTS compliance_blocked_record_id, - DROP COLUMN IF EXISTS status, - DROP COLUMN IF EXISTS article_version_id, - DROP COLUMN IF EXISTS article_id; + DROP CONSTRAINT IF EXISTS desktop_publish_jobs_compliance_blocked_record_id_fkey; ALTER TABLE article_versions DROP CONSTRAINT IF EXISTS fk_article_versions_manual_review; diff --git a/server/migrations/20260505102000_create_compliance_tables.up.sql b/server/migrations/20260505102000_create_compliance_tables.up.sql index 7f4cc3e..350921f 100644 --- a/server/migrations/20260505102000_create_compliance_tables.up.sql +++ b/server/migrations/20260505102000_create_compliance_tables.up.sql @@ -202,12 +202,13 @@ ALTER TABLE article_versions ON DELETE SET NULL; ALTER TABLE desktop_publish_jobs - ADD COLUMN IF NOT EXISTS article_id BIGINT REFERENCES articles(id), - ADD COLUMN IF NOT EXISTS article_version_id BIGINT REFERENCES article_versions(id), - ADD COLUMN IF NOT EXISTS status VARCHAR(32) NOT NULL DEFAULT 'queued', - ADD COLUMN IF NOT EXISTS compliance_blocked_record_id BIGINT REFERENCES compliance_check_records(id) ON DELETE SET NULL, - ADD COLUMN IF NOT EXISTS compliance_blocked_at TIMESTAMPTZ, - ADD COLUMN IF NOT EXISTS compliance_blocked_reason TEXT; + DROP CONSTRAINT IF EXISTS desktop_publish_jobs_compliance_blocked_record_id_fkey; + +ALTER TABLE desktop_publish_jobs + ADD CONSTRAINT desktop_publish_jobs_compliance_blocked_record_id_fkey + FOREIGN KEY (compliance_blocked_record_id) + REFERENCES compliance_check_records(id) + ON DELETE SET NULL; CREATE INDEX IF NOT EXISTS idx_desktop_publish_jobs_status_scheduled ON desktop_publish_jobs (status, scheduled_at, created_at)