diff --git a/apps/ops-web/src/layouts/AppShell.vue b/apps/ops-web/src/layouts/AppShell.vue index fb35cae..9a90fdf 100644 --- a/apps/ops-web/src/layouts/AppShell.vue +++ b/apps/ops-web/src/layouts/AppShell.vue @@ -49,6 +49,7 @@ import { AuditOutlined, BookOutlined, + ClockCircleOutlined, ControlOutlined, CrownOutlined, FileSearchOutlined, @@ -89,6 +90,7 @@ const menuLeaves: MenuLeaf[] = [ { key: '/compliance/stats', label: '合规统计', path: '/compliance/stats' }, { key: '/accounts', label: '操作员管理', path: '/accounts' }, { key: '/jobs', label: '任务中心', path: '/jobs' }, + { key: '/scheduler', label: '调度中心', path: '/scheduler' }, { key: '/audits', label: '审计日志', path: '/audits' }, ] @@ -155,6 +157,11 @@ const menuItems = computed(() => [ label: '任务中心', icon: () => h(PartitionOutlined), }, + { + key: '/scheduler', + label: '调度中心', + icon: () => h(ClockCircleOutlined), + }, { key: '/audits', label: '审计日志', diff --git a/apps/ops-web/src/router/index.ts b/apps/ops-web/src/router/index.ts index 0a370f0..b4ae0f5 100644 --- a/apps/ops-web/src/router/index.ts +++ b/apps/ops-web/src/router/index.ts @@ -49,6 +49,12 @@ export const router = createRouter({ component: () => import('@/views/JobsView.vue'), meta: { title: '任务中心' }, }, + { + path: 'scheduler', + name: 'scheduler', + component: () => import('@/views/SchedulerView.vue'), + meta: { title: '调度中心' }, + }, { path: 'site-domain-mappings', name: 'site-domain-mappings', diff --git a/apps/ops-web/src/views/SchedulerView.vue b/apps/ops-web/src/views/SchedulerView.vue new file mode 100644 index 0000000..1ec191c --- /dev/null +++ b/apps/ops-web/src/views/SchedulerView.vue @@ -0,0 +1,731 @@ + + + + + diff --git a/server/cmd/ops-api/main.go b/server/cmd/ops-api/main.go index a491e3b..444876c 100644 --- a/server/cmd/ops-api/main.go +++ b/server/cmd/ops-api/main.go @@ -83,6 +83,7 @@ func main() { kolSubscriptionsRepo := repository.NewKolSubscriptionRepository(pool) auditsRepo := repository.NewAuditRepository(pool) siteDomainMappingsRepo := repository.NewSiteDomainMappingRepository(monitoringPool) + schedulerRepo := repository.NewSchedulerRepository(pool) ipRegionResolver, err := ipregion.NewResolver(cfg.IPRegion.V4XDBPath, cfg.IPRegion.V6XDBPath, logger) if err != nil { @@ -118,6 +119,7 @@ func main() { kolSubscriptionSvc := app.NewKolSubscriptionService(kolSubscriptionsRepo, auditSvc).WithCache(appCache) siteDomainMappingSvc := app.NewSiteDomainMappingService(siteDomainMappingsRepo, auditSvc) complianceSvc := app.NewComplianceService(pool, auditSvc, logger) + schedulerSvc := app.NewSchedulerService(schedulerRepo, auditSvc) if err := app.SeedDefaultAdmin(ctx, accountsRepo, app.DefaultAdminSeed{ Username: cfg.DefaultAdmin.Username, @@ -165,6 +167,7 @@ func main() { Audits: auditSvc, SiteDomains: siteDomainMappingSvc, Compliance: complianceSvc, + Scheduler: schedulerSvc, }) addr := fmt.Sprintf(":%d", cfg.Server.Port) diff --git a/server/cmd/scheduler/main.go b/server/cmd/scheduler/main.go index 64efcb1..a1d1bf3 100644 --- a/server/cmd/scheduler/main.go +++ b/server/cmd/scheduler/main.go @@ -95,14 +95,59 @@ func main() { generationStateCheckWorker := generateworker.NewGenerationTaskStateCheckWorker(app.DB, app.Logger, app.Config().Generation). WithConfigProvider(app.ConfigStore). WithCache(app.Cache) + monitoringRetentionWorker := internalscheduler.NewMonitoringRetentionWorker(app.MonitoringDB, app.Logger) - startSchedulerWorker(ctx, &workerWG, "schedule_dispatch", app.Logger.Sugar(), scheduleDispatchWorker.Run) - startSchedulerWorker(ctx, &workerWG, "monitoring_daily_task", app.Logger.Sugar(), monitoringDailyTaskWorker.Run) - startSchedulerWorker(ctx, &workerWG, "monitoring_result_recovery", app.Logger.Sugar(), monitoringResultRecoveryWorker.Run) - startSchedulerWorker(ctx, &workerWG, "monitoring_lease_recovery", app.Logger.Sugar(), monitoringLeaseRecoveryWorker.Run) - startSchedulerWorker(ctx, &workerWG, "monitoring_received_inspection", app.Logger.Sugar(), monitoringReceivedInspectionWorker.Run) - startSchedulerWorker(ctx, &workerWG, "knowledge_deleted_cleanup", app.Logger.Sugar(), knowledgeDeletedCleanupWorker.Run) - startSchedulerWorker(ctx, &workerWG, "generation_state_check", app.Logger.Sugar(), generationStateCheckWorker.Run) + controlPlane := internalscheduler.NewControlPlane(app.DB, app.Logger, []internalscheduler.ControlledJob{ + { + Key: "schedule_dispatch", + Run: func(runCtx context.Context, _ internalscheduler.JobRunContext) (map[string]any, error) { + return scheduleDispatchWorker.RunOnce(runCtx) + }, + }, + { + Key: "monitoring_daily_task", + Run: func(runCtx context.Context, _ internalscheduler.JobRunContext) (map[string]any, error) { + return monitoringDailyTaskWorker.RunOnce(runCtx) + }, + }, + { + Key: "monitoring_result_recovery", + Run: func(runCtx context.Context, _ internalscheduler.JobRunContext) (map[string]any, error) { + return monitoringResultRecoveryWorker.RunOnce(runCtx) + }, + }, + { + Key: "monitoring_lease_recovery", + Run: func(runCtx context.Context, _ internalscheduler.JobRunContext) (map[string]any, error) { + return monitoringLeaseRecoveryWorker.RunOnce(runCtx) + }, + }, + { + Key: "monitoring_received_inspection", + Run: func(runCtx context.Context, _ internalscheduler.JobRunContext) (map[string]any, error) { + return monitoringReceivedInspectionWorker.RunOnce(runCtx) + }, + }, + { + Key: "knowledge_deleted_cleanup", + Run: func(runCtx context.Context, _ internalscheduler.JobRunContext) (map[string]any, error) { + return knowledgeDeletedCleanupWorker.RunOnce(runCtx) + }, + }, + { + Key: "generation_state_check", + Run: func(runCtx context.Context, _ internalscheduler.JobRunContext) (map[string]any, error) { + return generationStateCheckWorker.RunOnce(runCtx) + }, + }, + { + Key: "monitoring_retention_cleanup", + Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) { + return monitoringRetentionWorker.RunOnce(runCtx, jobCtx) + }, + }, + }) + startSchedulerWorker(ctx, &workerWG, "control_plane", app.Logger.Sugar(), controlPlane.Run) var metricsServer *http.Server if app.Config().Scheduler.HTTPPort > 0 { diff --git a/server/internal/ops/app/scheduler.go b/server/internal/ops/app/scheduler.go new file mode 100644 index 0000000..b7c1f4f --- /dev/null +++ b/server/internal/ops/app/scheduler.go @@ -0,0 +1,301 @@ +package app + +import ( + "context" + "errors" + "fmt" + "strings" + "time" + + "github.com/geo-platform/tenant-api/internal/ops/domain" + "github.com/geo-platform/tenant-api/internal/ops/repository" + "github.com/geo-platform/tenant-api/internal/shared/response" +) + +const ( + ActionSchedulerJobUpdate = "scheduler_job.update" + ActionSchedulerJobEnable = "scheduler_job.enable" + ActionSchedulerJobDisable = "scheduler_job.disable" + ActionSchedulerJobRun = "scheduler_job.run" + ActionSchedulerJobDryRun = "scheduler_job.dry_run" +) + +type SchedulerService struct { + repo *repository.SchedulerRepository + audits *AuditService +} + +func NewSchedulerService(repo *repository.SchedulerRepository, audits *AuditService) *SchedulerService { + return &SchedulerService{repo: repo, audits: audits} +} + +type SchedulerJobListInput struct { + Category string + Keyword string + Enabled *bool + Page int + Size int +} + +type SchedulerJobListResult struct { + Items []domain.SchedulerJob `json:"items"` + Total int64 `json:"total"` + Page int `json:"page"` + Size int `json:"size"` +} + +type SchedulerRunListResult struct { + Items []domain.SchedulerRun `json:"items"` + Total int64 `json:"total"` + Page int `json:"page"` + Size int `json:"size"` +} + +type SchedulerJobUpdateInput struct { + Enabled *bool `json:"enabled"` + ScheduleType *string `json:"schedule_type"` + IntervalSeconds *int `json:"interval_seconds"` + CronExpr *string `json:"cron_expr"` + Timezone *string `json:"timezone"` + TimeoutSeconds *int `json:"timeout_seconds"` + BatchSize *int `json:"batch_size"` + MaxConcurrency *int `json:"max_concurrency"` + Config map[string]any `json:"config"` +} + +type SchedulerTriggerInput struct { + Reason string `json:"reason"` + ConfigOverride map[string]any `json:"config_override"` +} + +func (s *SchedulerService) ListJobs(ctx context.Context, in SchedulerJobListInput) (*SchedulerJobListResult, error) { + page := in.Page + if page <= 0 { + page = 1 + } + size := in.Size + if size <= 0 || size > 200 { + size = 50 + } + items, total, err := s.repo.ListJobs(ctx, repository.SchedulerJobFilter{ + Category: strings.TrimSpace(in.Category), + Keyword: strings.TrimSpace(in.Keyword), + Enabled: in.Enabled, + Limit: size, + Offset: (page - 1) * size, + }) + if err != nil { + return nil, response.ErrInternal(53001, "scheduler_jobs_query_failed", "failed to list scheduler jobs") + } + return &SchedulerJobListResult{Items: items, Total: total, Page: page, Size: size}, nil +} + +func (s *SchedulerService) GetJob(ctx context.Context, key string) (*domain.SchedulerJob, error) { + item, err := s.repo.GetJob(ctx, normalizeSchedulerJobKey(key)) + if err != nil { + return nil, mapSchedulerJobError(err) + } + return item, nil +} + +func (s *SchedulerService) UpdateJob(ctx context.Context, actor *Actor, key string, in SchedulerJobUpdateInput) (*domain.SchedulerJob, error) { + normalized, err := normalizeSchedulerJobUpdate(in) + if err != nil { + return nil, err + } + updatedBy := actorID(actor) + item, err := s.repo.UpdateJob(ctx, normalizeSchedulerJobKey(key), repository.SchedulerJobUpdate{ + Enabled: normalized.Enabled, + ScheduleType: normalized.ScheduleType, + IntervalSeconds: normalized.IntervalSeconds, + CronExpr: normalized.CronExpr, + Timezone: normalized.Timezone, + TimeoutSeconds: normalized.TimeoutSeconds, + BatchSize: normalized.BatchSize, + MaxConcurrency: normalized.MaxConcurrency, + Config: normalized.Config, + UpdatedBy: updatedBy, + }) + if err != nil { + return nil, mapSchedulerJobError(err) + } + s.auditScheduler(ctx, actor, ActionSchedulerJobUpdate, item.JobKey, map[string]any{ + "version": item.Version, + }) + return item, nil +} + +func (s *SchedulerService) SetEnabled(ctx context.Context, actor *Actor, key string, enabled bool) (*domain.SchedulerJob, error) { + item, err := s.repo.SetJobEnabled(ctx, normalizeSchedulerJobKey(key), enabled, actorID(actor)) + if err != nil { + return nil, mapSchedulerJobError(err) + } + action := ActionSchedulerJobDisable + if enabled { + action = ActionSchedulerJobEnable + } + s.auditScheduler(ctx, actor, action, item.JobKey, map[string]any{"version": item.Version}) + return item, nil +} + +func (s *SchedulerService) Trigger(ctx context.Context, actor *Actor, key string, triggerType string, in SchedulerTriggerInput) (*domain.SchedulerTrigger, error) { + key = normalizeSchedulerJobKey(key) + if _, err := s.repo.GetJobForRuntime(ctx, key); err != nil { + return nil, mapSchedulerJobError(err) + } + triggerType = strings.TrimSpace(triggerType) + if triggerType != domain.SchedulerTriggerManual && triggerType != domain.SchedulerTriggerDryRun { + return nil, response.ErrBadRequest(40083, "invalid_trigger_type", "trigger_type 不合法") + } + trigger, err := s.repo.CreateTrigger(ctx, repository.SchedulerTriggerCreate{ + JobKey: key, + TriggerType: triggerType, + RequestedBy: actorID(actor), + Reason: strings.TrimSpace(in.Reason), + ConfigOverride: sanitizeSchedulerConfig(in.ConfigOverride), + }) + if err != nil { + return nil, response.ErrInternal(53004, "scheduler_trigger_create_failed", "failed to create scheduler trigger") + } + action := ActionSchedulerJobRun + if triggerType == domain.SchedulerTriggerDryRun { + action = ActionSchedulerJobDryRun + } + s.auditScheduler(ctx, actor, action, key, map[string]any{ + "trigger_id": trigger.ID, + "reason": strings.TrimSpace(in.Reason), + }) + return trigger, nil +} + +func (s *SchedulerService) ListRuns(ctx context.Context, key, status string, page, size int) (*SchedulerRunListResult, error) { + page = maxInt(page, 1) + if size <= 0 || size > 200 { + size = 50 + } + items, total, err := s.repo.ListRuns(ctx, repository.SchedulerRunFilter{ + JobKey: normalizeSchedulerJobKey(key), + Status: strings.TrimSpace(status), + Limit: size, + Offset: (page - 1) * size, + }) + if err != nil { + return nil, response.ErrInternal(53005, "scheduler_runs_query_failed", "failed to list scheduler runs") + } + return &SchedulerRunListResult{Items: items, Total: total, Page: page, Size: size}, nil +} + +func (s *SchedulerService) ListInstances(ctx context.Context) ([]domain.SchedulerInstance, error) { + items, err := s.repo.ListInstances(ctx) + if err != nil { + return nil, response.ErrInternal(53006, "scheduler_instances_query_failed", "failed to list scheduler instances") + } + return items, nil +} + +func normalizeSchedulerJobUpdate(in SchedulerJobUpdateInput) (SchedulerJobUpdateInput, error) { + out := in + if out.ScheduleType != nil { + value := strings.TrimSpace(*out.ScheduleType) + if value != "interval" && value != "cron" && value != "manual" { + return out, response.ErrBadRequest(40080, "invalid_schedule_type", "schedule_type 不合法") + } + out.ScheduleType = &value + } + if out.IntervalSeconds != nil && *out.IntervalSeconds <= 0 { + return out, response.ErrBadRequest(40081, "invalid_interval_seconds", "interval_seconds 必须大于 0") + } + if out.TimeoutSeconds != nil && *out.TimeoutSeconds <= 0 { + return out, response.ErrBadRequest(40082, "invalid_timeout_seconds", "timeout_seconds 必须大于 0") + } + if out.BatchSize != nil && *out.BatchSize <= 0 { + return out, response.ErrBadRequest(40084, "invalid_batch_size", "batch_size 必须大于 0") + } + if out.MaxConcurrency != nil && *out.MaxConcurrency <= 0 { + return out, response.ErrBadRequest(40085, "invalid_max_concurrency", "max_concurrency 必须大于 0") + } + if out.Timezone != nil { + value := strings.TrimSpace(*out.Timezone) + if value == "" { + value = "Asia/Shanghai" + } + if _, err := time.LoadLocation(value); err != nil { + return out, response.ErrBadRequest(40086, "invalid_timezone", "timezone 不合法") + } + out.Timezone = &value + } + if out.CronExpr != nil { + value := strings.TrimSpace(*out.CronExpr) + out.CronExpr = &value + } + out.Config = sanitizeSchedulerConfig(out.Config) + return out, nil +} + +func sanitizeSchedulerConfig(config map[string]any) map[string]any { + if config == nil { + return nil + } + out := make(map[string]any, len(config)) + for key, value := range config { + trimmed := strings.TrimSpace(key) + if trimmed == "" { + continue + } + out[trimmed] = value + } + return out +} + +func normalizeSchedulerJobKey(key string) string { + return strings.TrimSpace(key) +} + +func mapSchedulerJobError(err error) error { + if errors.Is(err, repository.ErrSchedulerJobNotFound) { + return response.ErrNotFound(40480, "scheduler_job_not_found", "调度任务不存在") + } + return response.ErrInternal(53002, "scheduler_job_query_failed", "failed to load scheduler job") +} + +func actorID(actor *Actor) *int64 { + if actor == nil || actor.OperatorID <= 0 { + return nil + } + id := actor.OperatorID + return &id +} + +func (s *SchedulerService) auditScheduler(ctx context.Context, actor *Actor, action, key string, metadata map[string]any) { + if actor == nil || s.audits == nil { + return + } + id := actor.OperatorID + event := domain.AuditEvent{ + OperatorID: &id, + OperatorName: actor.DisplayName, + Action: action, + TargetType: "scheduler_job", + TargetID: key, + Metadata: metadata, + IP: actor.IP, + UserAgent: actor.UserAgent, + RequestID: actor.RequestID, + } + _ = s.audits.Append(ctx, event) +} + +func maxInt(a, b int) int { + if a > b { + return a + } + return b +} + +func SchedulerJobConfigAsString(config map[string]any, key string) string { + value, ok := config[key] + if !ok || value == nil { + return "" + } + return fmt.Sprint(value) +} diff --git a/server/internal/ops/domain/scheduler.go b/server/internal/ops/domain/scheduler.go index e9290eb..1c4d2eb 100644 --- a/server/internal/ops/domain/scheduler.go +++ b/server/internal/ops/domain/scheduler.go @@ -37,19 +37,19 @@ type SchedulerJob struct { } type SchedulerRun struct { - ID int64 `json:"id"` - JobKey string `json:"job_key"` - TriggerID *int64 `json:"trigger_id,omitempty"` - SchedulerInstanceID *string `json:"scheduler_instance_id,omitempty"` - TriggerType string `json:"trigger_type"` - Status string `json:"status"` - ConfigVersion int `json:"config_version"` - ConfigSnapshot map[string]any `json:"config_snapshot"` - Stats map[string]any `json:"stats"` - ErrorMessage *string `json:"error_message,omitempty"` - StartedAt time.Time `json:"started_at"` - FinishedAt *time.Time `json:"finished_at,omitempty"` - DurationMilliseconds *int64 `json:"duration_ms,omitempty"` + ID int64 `json:"id"` + JobKey string `json:"job_key"` + TriggerID *int64 `json:"trigger_id,omitempty"` + SchedulerInstanceID *string `json:"scheduler_instance_id,omitempty"` + TriggerType string `json:"trigger_type"` + Status string `json:"status"` + ConfigVersion int `json:"config_version"` + ConfigSnapshot map[string]any `json:"config_snapshot"` + Stats map[string]any `json:"stats"` + ErrorMessage *string `json:"error_message,omitempty"` + StartedAt time.Time `json:"started_at"` + FinishedAt *time.Time `json:"finished_at,omitempty"` + DurationMilliseconds *int64 `json:"duration_ms,omitempty"` } type SchedulerInstance struct { diff --git a/server/internal/ops/repository/scheduler.go b/server/internal/ops/repository/scheduler.go new file mode 100644 index 0000000..00c12dc --- /dev/null +++ b/server/internal/ops/repository/scheduler.go @@ -0,0 +1,603 @@ +package repository + +import ( + "context" + "encoding/json" + "errors" + "fmt" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/geo-platform/tenant-api/internal/ops/domain" +) + +var ErrSchedulerJobNotFound = errors.New("scheduler job not found") + +type SchedulerRepository struct { + pool *pgxpool.Pool +} + +func NewSchedulerRepository(pool *pgxpool.Pool) *SchedulerRepository { + return &SchedulerRepository{pool: pool} +} + +type SchedulerJobFilter struct { + Category string + Keyword string + Enabled *bool + Limit int + Offset int +} + +type SchedulerJobUpdate struct { + Enabled *bool + ScheduleType *string + IntervalSeconds *int + CronExpr *string + Timezone *string + TimeoutSeconds *int + BatchSize *int + MaxConcurrency *int + Config map[string]any + UpdatedBy *int64 +} + +type SchedulerRunFilter struct { + JobKey string + Status string + Limit int + Offset int +} + +type SchedulerRunStart struct { + JobKey string + TriggerID *int64 + SchedulerInstanceID string + TriggerType string + ConfigVersion int + ConfigSnapshot map[string]any +} + +type SchedulerRunFinish struct { + RunID int64 + Status string + Stats map[string]any + ErrorMessage *string +} + +type SchedulerTriggerCreate struct { + JobKey string + TriggerType string + RequestedBy *int64 + Reason string + ConfigOverride map[string]any +} + +type SchedulerTriggerClaim struct { + ID int64 + JobKey string + TriggerType string + ConfigOverride map[string]any +} + +const schedulerJobSelect = ` + SELECT + j.job_key, + j.display_name, + j.category, + j.description, + j.enabled, + j.schedule_type, + j.interval_seconds, + j.cron_expr, + j.timezone, + j.timeout_seconds, + j.batch_size, + j.max_concurrency, + j.config, + j.version, + j.updated_by, + j.updated_at, + j.created_at, + COALESCE(pending.pending_count, 0) + FROM ops.scheduler_jobs j + LEFT JOIN LATERAL ( + SELECT COUNT(*)::int AS pending_count + FROM ops.scheduler_job_triggers t + WHERE t.job_key = j.job_key AND t.status = 'pending' + ) pending ON true +` + +func (r *SchedulerRepository) ListJobs(ctx context.Context, f SchedulerJobFilter) ([]domain.SchedulerJob, int64, error) { + args := []any{} + where := "1=1" + if f.Category != "" { + args = append(args, f.Category) + where += fmt.Sprintf(" AND j.category = $%d", len(args)) + } + if f.Keyword != "" { + args = append(args, "%"+f.Keyword+"%") + where += fmt.Sprintf(" AND (j.job_key ILIKE $%d OR j.display_name ILIKE $%d OR COALESCE(j.description, '') ILIKE $%d)", len(args), len(args), len(args)) + } + if f.Enabled != nil { + args = append(args, *f.Enabled) + where += fmt.Sprintf(" AND j.enabled = $%d", len(args)) + } + + var total int64 + if err := r.pool.QueryRow(ctx, "SELECT COUNT(*) FROM ops.scheduler_jobs j WHERE "+where, args...).Scan(&total); err != nil { + return nil, 0, err + } + + limit := f.Limit + if limit <= 0 || limit > 200 { + limit = 50 + } + offset := f.Offset + if offset < 0 { + offset = 0 + } + args = append(args, limit, offset) + limitArg := len(args) - 1 + offsetArg := len(args) + + rows, err := r.pool.Query(ctx, schedulerJobSelect+fmt.Sprintf(` + WHERE %s + ORDER BY j.category ASC, j.job_key ASC + LIMIT $%d OFFSET $%d`, where, limitArg, offsetArg), args...) + if err != nil { + return nil, 0, err + } + defer rows.Close() + + items := make([]domain.SchedulerJob, 0, limit) + for rows.Next() { + item, scanErr := scanSchedulerJob(rows) + if scanErr != nil { + return nil, 0, scanErr + } + items = append(items, *item) + } + if err := rows.Err(); err != nil { + return nil, 0, err + } + + if err := r.attachRunSnapshots(ctx, items); err != nil { + return nil, 0, err + } + return items, total, nil +} + +func (r *SchedulerRepository) GetJob(ctx context.Context, key string) (*domain.SchedulerJob, error) { + item, err := scanSchedulerJob(r.pool.QueryRow(ctx, schedulerJobSelect+` + WHERE j.job_key = $1`, key)) + if err != nil { + return nil, err + } + items := []domain.SchedulerJob{*item} + if err := r.attachRunSnapshots(ctx, items); err != nil { + return nil, err + } + *item = items[0] + return item, nil +} + +func (r *SchedulerRepository) GetJobForRuntime(ctx context.Context, key string) (*domain.SchedulerJob, error) { + return scanSchedulerJob(r.pool.QueryRow(ctx, schedulerJobSelect+` + WHERE j.job_key = $1`, key)) +} + +func (r *SchedulerRepository) UpdateJob(ctx context.Context, key string, in SchedulerJobUpdate) (*domain.SchedulerJob, error) { + configJSON, err := marshalSchedulerJSONMap(in.Config) + if err != nil { + return nil, err + } + return scanSchedulerJob(r.pool.QueryRow(ctx, ` + UPDATE ops.scheduler_jobs + SET enabled = COALESCE($2, enabled), + schedule_type = COALESCE(NULLIF($3, ''), schedule_type), + interval_seconds = COALESCE($4, interval_seconds), + cron_expr = COALESCE($5, cron_expr), + timezone = COALESCE(NULLIF($6, ''), timezone), + timeout_seconds = COALESCE($7, timeout_seconds), + batch_size = COALESCE($8, batch_size), + max_concurrency = COALESCE($9, max_concurrency), + config = COALESCE($10::jsonb, config), + updated_by = COALESCE($11, updated_by), + version = version + 1, + updated_at = NOW() + WHERE job_key = $1 + RETURNING job_key, display_name, category, description, enabled, schedule_type, + interval_seconds, cron_expr, timezone, timeout_seconds, batch_size, + max_concurrency, config, version, updated_by, updated_at, created_at, 0::int + `, key, in.Enabled, nullableStringPtr(in.ScheduleType), in.IntervalSeconds, in.CronExpr, + nullableStringPtr(in.Timezone), in.TimeoutSeconds, in.BatchSize, in.MaxConcurrency, configJSON, in.UpdatedBy)) +} + +func (r *SchedulerRepository) SetJobEnabled(ctx context.Context, key string, enabled bool, updatedBy *int64) (*domain.SchedulerJob, error) { + return r.UpdateJob(ctx, key, SchedulerJobUpdate{ + Enabled: &enabled, + UpdatedBy: updatedBy, + }) +} + +func (r *SchedulerRepository) CreateTrigger(ctx context.Context, in SchedulerTriggerCreate) (*domain.SchedulerTrigger, error) { + configJSON, err := marshalSchedulerJSONMap(in.ConfigOverride) + if err != nil { + return nil, err + } + reason := nullableString(in.Reason) + return scanSchedulerTrigger(r.pool.QueryRow(ctx, ` + INSERT INTO ops.scheduler_job_triggers (job_key, trigger_type, requested_by, reason, config_override) + VALUES ($1, $2, $3, $4, COALESCE($5::jsonb, '{}'::jsonb)) + RETURNING id, job_key, trigger_type, status, requested_by, reason, config_override, + scheduler_instance_id, run_id, requested_at, claimed_at, finished_at, error_message + `, in.JobKey, in.TriggerType, in.RequestedBy, reason, configJSON)) +} + +func (r *SchedulerRepository) ClaimPendingTrigger(ctx context.Context, jobKey, instanceID string) (*SchedulerTriggerClaim, bool, error) { + var claim SchedulerTriggerClaim + var configRaw []byte + err := r.pool.QueryRow(ctx, ` + UPDATE ops.scheduler_job_triggers t + SET status = 'claimed', + claimed_at = NOW(), + scheduler_instance_id = $2 + WHERE t.id = ( + SELECT id + FROM ops.scheduler_job_triggers + WHERE job_key = $1 AND status = 'pending' + ORDER BY requested_at ASC, id ASC + FOR UPDATE SKIP LOCKED + LIMIT 1 + ) + RETURNING id, job_key, trigger_type, config_override + `, jobKey, instanceID).Scan(&claim.ID, &claim.JobKey, &claim.TriggerType, &configRaw) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, false, nil + } + return nil, false, err + } + claim.ConfigOverride = decodeSchedulerJSONMap(configRaw) + return &claim, true, nil +} + +func (r *SchedulerRepository) FinishTrigger(ctx context.Context, triggerID int64, runID *int64, status string, errorMessage *string) error { + _, err := r.pool.Exec(ctx, ` + UPDATE ops.scheduler_job_triggers + SET status = $3, + run_id = $2, + finished_at = NOW(), + error_message = $4 + WHERE id = $1 + `, triggerID, runID, status, errorMessage) + return err +} + +func (r *SchedulerRepository) StartRun(ctx context.Context, in SchedulerRunStart) (*domain.SchedulerRun, error) { + configJSON, err := marshalSchedulerJSONMap(in.ConfigSnapshot) + if err != nil { + return nil, err + } + return scanSchedulerRun(r.pool.QueryRow(ctx, ` + INSERT INTO ops.scheduler_job_runs ( + job_key, trigger_id, scheduler_instance_id, trigger_type, + status, config_version, config_snapshot, started_at + ) + VALUES ($1, $2, $3, $4, 'running', $5, COALESCE($6::jsonb, '{}'::jsonb), NOW()) + RETURNING id, job_key, trigger_id, scheduler_instance_id, trigger_type, status, + config_version, config_snapshot, stats, error_message, started_at, + finished_at, duration_ms + `, in.JobKey, in.TriggerID, nullableString(in.SchedulerInstanceID), in.TriggerType, in.ConfigVersion, configJSON)) +} + +func (r *SchedulerRepository) FinishRun(ctx context.Context, in SchedulerRunFinish) (*domain.SchedulerRun, error) { + statsJSON, err := marshalSchedulerJSONMap(in.Stats) + if err != nil { + return nil, err + } + return scanSchedulerRun(r.pool.QueryRow(ctx, ` + UPDATE ops.scheduler_job_runs + SET status = $2, + stats = COALESCE($3::jsonb, '{}'::jsonb), + error_message = $4, + finished_at = NOW(), + duration_ms = GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (NOW() - started_at)) * 1000)::bigint) + WHERE id = $1 + RETURNING id, job_key, trigger_id, scheduler_instance_id, trigger_type, status, + config_version, config_snapshot, stats, error_message, started_at, + finished_at, duration_ms + `, in.RunID, in.Status, statsJSON, in.ErrorMessage)) +} + +func (r *SchedulerRepository) ListRuns(ctx context.Context, f SchedulerRunFilter) ([]domain.SchedulerRun, int64, error) { + args := []any{} + where := "1=1" + if f.JobKey != "" { + args = append(args, f.JobKey) + where += fmt.Sprintf(" AND job_key = $%d", len(args)) + } + if f.Status != "" { + args = append(args, f.Status) + where += fmt.Sprintf(" AND status = $%d", len(args)) + } + var total int64 + if err := r.pool.QueryRow(ctx, "SELECT COUNT(*) FROM ops.scheduler_job_runs WHERE "+where, args...).Scan(&total); err != nil { + return nil, 0, err + } + limit := f.Limit + if limit <= 0 || limit > 200 { + limit = 50 + } + offset := f.Offset + if offset < 0 { + offset = 0 + } + args = append(args, limit, offset) + limitArg := len(args) - 1 + offsetArg := len(args) + rows, err := r.pool.Query(ctx, fmt.Sprintf(` + SELECT id, job_key, trigger_id, scheduler_instance_id, trigger_type, status, + config_version, config_snapshot, stats, error_message, started_at, + finished_at, duration_ms + FROM ops.scheduler_job_runs + WHERE %s + ORDER BY started_at DESC, id DESC + LIMIT $%d OFFSET $%d + `, where, limitArg, offsetArg), args...) + if err != nil { + return nil, 0, err + } + defer rows.Close() + items := make([]domain.SchedulerRun, 0, limit) + for rows.Next() { + item, scanErr := scanSchedulerRun(rows) + if scanErr != nil { + return nil, 0, scanErr + } + items = append(items, *item) + } + return items, total, rows.Err() +} + +func (r *SchedulerRepository) UpsertInstance(ctx context.Context, instance domain.SchedulerInstance) error { + metadataJSON, err := marshalSchedulerJSONMap(instance.Metadata) + if err != nil { + return err + } + _, err = r.pool.Exec(ctx, ` + INSERT INTO ops.scheduler_instances (instance_id, hostname, process, build_version, metadata, started_at, last_seen_at) + VALUES ($1, $2, $3, $4, COALESCE($5::jsonb, '{}'::jsonb), $6, NOW()) + ON CONFLICT (instance_id) DO UPDATE + SET hostname = EXCLUDED.hostname, + process = EXCLUDED.process, + build_version = EXCLUDED.build_version, + metadata = EXCLUDED.metadata, + last_seen_at = NOW() + `, instance.InstanceID, instance.Hostname, instance.Process, instance.BuildVersion, metadataJSON, instance.StartedAt) + return err +} + +func (r *SchedulerRepository) ListInstances(ctx context.Context) ([]domain.SchedulerInstance, error) { + rows, err := r.pool.Query(ctx, ` + SELECT instance_id, hostname, process, build_version, started_at, last_seen_at, metadata + FROM ops.scheduler_instances + ORDER BY last_seen_at DESC + LIMIT 200 + `) + if err != nil { + return nil, err + } + defer rows.Close() + items := make([]domain.SchedulerInstance, 0, 16) + for rows.Next() { + item, scanErr := scanSchedulerInstance(rows) + if scanErr != nil { + return nil, scanErr + } + items = append(items, *item) + } + return items, rows.Err() +} + +func (r *SchedulerRepository) attachRunSnapshots(ctx context.Context, items []domain.SchedulerJob) error { + if len(items) == 0 { + return nil + } + keys := make([]string, 0, len(items)) + for _, item := range items { + keys = append(keys, item.JobKey) + } + rows, err := r.pool.Query(ctx, ` + WITH ranked AS ( + SELECT + id, job_key, trigger_id, scheduler_instance_id, trigger_type, status, + config_version, config_snapshot, stats, error_message, started_at, + finished_at, duration_ms, + ROW_NUMBER() OVER (PARTITION BY job_key ORDER BY started_at DESC, id DESC) AS rn + FROM ops.scheduler_job_runs + WHERE job_key = ANY($1) + ) + SELECT id, job_key, trigger_id, scheduler_instance_id, trigger_type, status, + config_version, config_snapshot, stats, error_message, started_at, + finished_at, duration_ms + FROM ranked + WHERE rn = 1 + `, keys) + if err != nil { + return err + } + defer rows.Close() + lastRuns := map[string]*domain.SchedulerRun{} + for rows.Next() { + run, scanErr := scanSchedulerRun(rows) + if scanErr != nil { + return scanErr + } + lastRuns[run.JobKey] = run + } + if err := rows.Err(); err != nil { + return err + } + + runningRows, err := r.pool.Query(ctx, ` + SELECT DISTINCT ON (job_key) + id, job_key, trigger_id, scheduler_instance_id, trigger_type, status, + config_version, config_snapshot, stats, error_message, started_at, + finished_at, duration_ms + FROM ops.scheduler_job_runs + WHERE job_key = ANY($1) AND status = 'running' + ORDER BY job_key, started_at DESC, id DESC + `, keys) + if err != nil { + return err + } + defer runningRows.Close() + runningRuns := map[string]*domain.SchedulerRun{} + for runningRows.Next() { + run, scanErr := scanSchedulerRun(runningRows) + if scanErr != nil { + return scanErr + } + runningRuns[run.JobKey] = run + } + if err := runningRows.Err(); err != nil { + return err + } + + for i := range items { + items[i].LastRun = lastRuns[items[i].JobKey] + items[i].RunningRun = runningRuns[items[i].JobKey] + } + return nil +} + +func scanSchedulerJob(row pgx.Row) (*domain.SchedulerJob, error) { + var item domain.SchedulerJob + var configRaw []byte + err := row.Scan( + &item.JobKey, + &item.DisplayName, + &item.Category, + &item.Description, + &item.Enabled, + &item.ScheduleType, + &item.IntervalSeconds, + &item.CronExpr, + &item.Timezone, + &item.TimeoutSeconds, + &item.BatchSize, + &item.MaxConcurrency, + &configRaw, + &item.Version, + &item.UpdatedBy, + &item.UpdatedAt, + &item.CreatedAt, + &item.PendingTriggers, + ) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrSchedulerJobNotFound + } + return nil, err + } + item.Config = decodeSchedulerJSONMap(configRaw) + return &item, nil +} + +func scanSchedulerRun(row pgx.Row) (*domain.SchedulerRun, error) { + var item domain.SchedulerRun + var configRaw, statsRaw []byte + err := row.Scan( + &item.ID, + &item.JobKey, + &item.TriggerID, + &item.SchedulerInstanceID, + &item.TriggerType, + &item.Status, + &item.ConfigVersion, + &configRaw, + &statsRaw, + &item.ErrorMessage, + &item.StartedAt, + &item.FinishedAt, + &item.DurationMilliseconds, + ) + if err != nil { + return nil, err + } + item.ConfigSnapshot = decodeSchedulerJSONMap(configRaw) + item.Stats = decodeSchedulerJSONMap(statsRaw) + return &item, nil +} + +func scanSchedulerTrigger(row pgx.Row) (*domain.SchedulerTrigger, error) { + var item domain.SchedulerTrigger + var configRaw []byte + err := row.Scan( + &item.ID, + &item.JobKey, + &item.TriggerType, + &item.Status, + &item.RequestedBy, + &item.Reason, + &configRaw, + &item.SchedulerInstanceID, + &item.RunID, + &item.RequestedAt, + &item.ClaimedAt, + &item.FinishedAt, + &item.ErrorMessage, + ) + if err != nil { + return nil, err + } + item.ConfigOverride = decodeSchedulerJSONMap(configRaw) + return &item, nil +} + +func scanSchedulerInstance(row pgx.Row) (*domain.SchedulerInstance, error) { + var item domain.SchedulerInstance + var metadataRaw []byte + if err := row.Scan( + &item.InstanceID, + &item.Hostname, + &item.Process, + &item.BuildVersion, + &item.StartedAt, + &item.LastSeenAt, + &metadataRaw, + ); err != nil { + return nil, err + } + item.Metadata = decodeSchedulerJSONMap(metadataRaw) + return &item, nil +} + +func marshalSchedulerJSONMap(value map[string]any) ([]byte, error) { + if value == nil { + return nil, nil + } + return json.Marshal(value) +} + +func decodeSchedulerJSONMap(raw []byte) map[string]any { + if len(raw) == 0 { + return map[string]any{} + } + var out map[string]any + if err := json.Unmarshal(raw, &out); err != nil || out == nil { + return map[string]any{} + } + return out +} + +func nullableStringPtr(value *string) *string { + if value == nil { + return nil + } + trimmed := *value + return &trimmed +} diff --git a/server/internal/ops/transport/router.go b/server/internal/ops/transport/router.go index 46c3107..9ebe33b 100644 --- a/server/internal/ops/transport/router.go +++ b/server/internal/ops/transport/router.go @@ -26,6 +26,7 @@ type Deps struct { Audits *app.AuditService SiteDomains *app.SiteDomainMappingService Compliance *app.ComplianceService + Scheduler *app.SchedulerService } func (d Deps) ServerAllowedOrigins() []string { @@ -115,6 +116,16 @@ func RegisterRoutes(d Deps) { authed.POST("/jobs/:source/:id/retry", retryJobHandler(d.Jobs)) authed.POST("/jobs/:source/:id/cancel", cancelJobHandler(d.Jobs)) + authed.GET("/scheduler/jobs", listSchedulerJobsHandler(d.Scheduler)) + authed.GET("/scheduler/jobs/:key", getSchedulerJobHandler(d.Scheduler)) + authed.PATCH("/scheduler/jobs/:key", updateSchedulerJobHandler(d.Scheduler)) + authed.POST("/scheduler/jobs/:key/enable", enableSchedulerJobHandler(d.Scheduler, true)) + authed.POST("/scheduler/jobs/:key/disable", enableSchedulerJobHandler(d.Scheduler, false)) + authed.POST("/scheduler/jobs/:key/run", triggerSchedulerJobHandler(d.Scheduler, "manual")) + authed.POST("/scheduler/jobs/:key/dry-run", triggerSchedulerJobHandler(d.Scheduler, "dry_run")) + authed.GET("/scheduler/jobs/:key/runs", listSchedulerRunsHandler(d.Scheduler)) + authed.GET("/scheduler/instances", listSchedulerInstancesHandler(d.Scheduler)) + 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/internal/ops/transport/scheduler_handler.go b/server/internal/ops/transport/scheduler_handler.go new file mode 100644 index 0000000..a6d47e0 --- /dev/null +++ b/server/internal/ops/transport/scheduler_handler.go @@ -0,0 +1,121 @@ +package transport + +import ( + "errors" + "io" + "strconv" + "strings" + + "github.com/gin-gonic/gin" + + "github.com/geo-platform/tenant-api/internal/ops/app" + "github.com/geo-platform/tenant-api/internal/shared/response" +) + +func listSchedulerJobsHandler(svc *app.SchedulerService) gin.HandlerFunc { + return func(c *gin.Context) { + page, _ := strconv.Atoi(c.DefaultQuery("page", "1")) + size, _ := strconv.Atoi(c.DefaultQuery("size", "50")) + var enabled *bool + if raw := strings.TrimSpace(c.Query("enabled")); raw != "" { + value, err := strconv.ParseBool(raw) + if err != nil { + response.Error(c, response.ErrBadRequest(40087, "invalid_enabled", "enabled 必须是布尔值")) + return + } + enabled = &value + } + result, err := svc.ListJobs(c.Request.Context(), app.SchedulerJobListInput{ + Category: strings.TrimSpace(c.Query("category")), + Keyword: strings.TrimSpace(c.Query("keyword")), + Enabled: enabled, + Page: page, + Size: size, + }) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, result) + } +} + +func getSchedulerJobHandler(svc *app.SchedulerService) gin.HandlerFunc { + return func(c *gin.Context) { + item, err := svc.GetJob(c.Request.Context(), c.Param("key")) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, item) + } +} + +func updateSchedulerJobHandler(svc *app.SchedulerService) gin.HandlerFunc { + return func(c *gin.Context) { + var req app.SchedulerJobUpdateInput + if err := c.ShouldBindJSON(&req); err != nil { + response.Error(c, response.ErrBadRequest(40088, "invalid_payload", err.Error())) + return + } + item, err := svc.UpdateJob(c.Request.Context(), actorFromGin(c), c.Param("key"), req) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, item) + } +} + +func enableSchedulerJobHandler(svc *app.SchedulerService, enabled bool) gin.HandlerFunc { + return func(c *gin.Context) { + item, err := svc.SetEnabled(c.Request.Context(), actorFromGin(c), c.Param("key"), enabled) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, item) + } +} + +func triggerSchedulerJobHandler(svc *app.SchedulerService, triggerType string) gin.HandlerFunc { + return func(c *gin.Context) { + var req app.SchedulerTriggerInput + if c.Request.Body != nil { + if err := c.ShouldBindJSON(&req); err != nil && !errors.Is(err, io.EOF) { + response.Error(c, response.ErrBadRequest(40089, "invalid_payload", err.Error())) + return + } + } + item, err := svc.Trigger(c.Request.Context(), actorFromGin(c), c.Param("key"), triggerType, req) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, item) + } +} + +func listSchedulerRunsHandler(svc *app.SchedulerService) gin.HandlerFunc { + return func(c *gin.Context) { + page, _ := strconv.Atoi(c.DefaultQuery("page", "1")) + size, _ := strconv.Atoi(c.DefaultQuery("size", "50")) + result, err := svc.ListRuns(c.Request.Context(), c.Param("key"), strings.TrimSpace(c.Query("status")), page, size) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, result) + } +} + +func listSchedulerInstancesHandler(svc *app.SchedulerService) gin.HandlerFunc { + return func(c *gin.Context) { + items, err := svc.ListInstances(c.Request.Context()) + if err != nil { + response.Error(c, err) + return + } + response.Success(c, gin.H{"items": items}) + } +} diff --git a/server/internal/scheduler/control_plane.go b/server/internal/scheduler/control_plane.go new file mode 100644 index 0000000..8406015 --- /dev/null +++ b/server/internal/scheduler/control_plane.go @@ -0,0 +1,498 @@ +package scheduler + +import ( + "context" + "errors" + "fmt" + "hash/fnv" + "os" + "strconv" + "strings" + "sync" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" + + opsdomain "github.com/geo-platform/tenant-api/internal/ops/domain" + opsrepo "github.com/geo-platform/tenant-api/internal/ops/repository" + sharedschedule "github.com/geo-platform/tenant-api/internal/shared/schedule" +) + +type ControlledJob struct { + Key string + DefaultName string + Run func(context.Context, JobRunContext) (map[string]any, error) +} + +type JobRunContext struct { + Job *opsdomain.SchedulerJob + TriggerID *int64 + Trigger string + DryRun bool + Config map[string]any +} + +type ControlPlane struct { + pool *pgxpool.Pool + repo *opsrepo.SchedulerRepository + logger *zap.Logger + instanceID string + hostname string + startedAt time.Time + jobs map[string]ControlledJob +} + +func NewControlPlane(pool *pgxpool.Pool, logger *zap.Logger, jobs []ControlledJob) *ControlPlane { + hostname, _ := os.Hostname() + if hostname == "" { + hostname = "unknown" + } + instanceID := hostname + ":" + strconv.Itoa(os.Getpid()) + ":" + strconv.FormatInt(time.Now().UnixNano(), 36) + byKey := make(map[string]ControlledJob, len(jobs)) + for _, job := range jobs { + if job.Key == "" || job.Run == nil { + continue + } + byKey[job.Key] = job + } + return &ControlPlane{ + pool: pool, + repo: opsrepo.NewSchedulerRepository(pool), + logger: logger, + instanceID: instanceID, + hostname: hostname, + startedAt: time.Now().UTC(), + jobs: byKey, + } +} + +func (p *ControlPlane) Run(ctx context.Context) { + if p == nil || p.pool == nil || len(p.jobs) == 0 { + return + } + + p.heartbeat(context.Background()) + heartbeatCtx, heartbeatCancel := context.WithCancel(ctx) + defer heartbeatCancel() + go p.runHeartbeat(heartbeatCtx) + + var wg sync.WaitGroup + for _, job := range p.jobs { + job := job + wg.Add(1) + go func() { + defer wg.Done() + p.runJobLoop(ctx, job) + }() + } + wg.Wait() +} + +func (p *ControlPlane) InstanceID() string { + if p == nil { + return "" + } + return p.instanceID +} + +func (p *ControlPlane) runHeartbeat(ctx context.Context) { + ticker := time.NewTicker(15 * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + p.heartbeat(context.Background()) + } + } +} + +func (p *ControlPlane) heartbeat(ctx context.Context) { + if p == nil || p.repo == nil { + return + } + buildVersion := os.Getenv("BUILD_VERSION") + if buildVersion == "" { + buildVersion = os.Getenv("GIT_SHA") + } + var buildVersionPtr *string + if buildVersion != "" { + buildVersionPtr = &buildVersion + } + if err := p.repo.UpsertInstance(ctx, opsdomain.SchedulerInstance{ + InstanceID: p.instanceID, + Hostname: p.hostname, + Process: "scheduler", + BuildVersion: buildVersionPtr, + StartedAt: p.startedAt, + LastSeenAt: time.Now().UTC(), + Metadata: map[string]any{ + "pid": os.Getpid(), + }, + }); err != nil && p.logger != nil { + p.logger.Warn("scheduler control heartbeat failed", zap.Error(err)) + } +} + +func (p *ControlPlane) runJobLoop(ctx context.Context, job ControlledJob) { + nextAutoAt := p.nextAutoAt(context.Background(), job.Key, time.Now(), true) + + for { + p.drainManualTriggers(context.Background(), job) + now := time.Now() + if !now.Before(nextAutoAt) { + p.runJobOnce(context.Background(), job, opsdomain.SchedulerTriggerAuto, nil, nil) + nextAutoAt = p.nextAutoAt(context.Background(), job.Key, time.Now(), false) + } + sleepFor := time.Until(nextAutoAt) + if sleepFor > 15*time.Second { + sleepFor = 15 * time.Second + } + if sleepFor <= 0 { + sleepFor = time.Second + } + timer := time.NewTimer(sleepFor) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + p.drainManualTriggers(context.Background(), job) + p.runJobOnce(context.Background(), job, opsdomain.SchedulerTriggerAuto, nil, nil) + } + } +} + +func (p *ControlPlane) drainManualTriggers(parent context.Context, job ControlledJob) { + for { + trigger, ok, err := p.repo.ClaimPendingTrigger(parent, job.Key, p.instanceID) + if err != nil { + if p.logger != nil { + p.logger.Warn("scheduler trigger claim failed", zap.String("job_key", job.Key), zap.Error(err)) + } + return + } + if !ok { + return + } + triggerID := trigger.ID + p.runJobOnce(parent, job, trigger.TriggerType, &triggerID, trigger.ConfigOverride) + } +} + +func (p *ControlPlane) nextAutoAt(ctx context.Context, key string, now time.Time, initial bool) time.Time { + job, err := p.repo.GetJobForRuntime(ctx, key) + if err != nil || job == nil { + return now.Add(30 * time.Second) + } + if job.ScheduleType == "manual" { + return now.Add(15 * time.Second) + } + if job.ScheduleType == "cron" && job.CronExpr != nil { + loc, locErr := time.LoadLocation(job.Timezone) + if locErr != nil { + loc = time.Local + } + schedule, parseErr := sharedschedule.Parse(*job.CronExpr) + if parseErr != nil { + if p.logger != nil { + p.logger.Warn("scheduler controlled job has invalid cron", + zap.String("job_key", key), + zap.String("cron_expr", *job.CronExpr), + zap.Error(parseErr), + ) + } + return now.Add(30 * time.Second) + } + base := now.In(loc) + return schedule.Next(base) + } + if runAt := parseRunAtLocal(job.Config); runAt != "" { + loc, locErr := time.LoadLocation(job.Timezone) + if locErr != nil { + loc = time.Local + } + hour, minute, ok := parseHourMinute(runAt) + if ok { + localNow := now.In(loc) + candidate := time.Date(localNow.Year(), localNow.Month(), localNow.Day(), hour, minute, 0, 0, loc) + if !initial || !localNow.Before(candidate) { + candidate = candidate.AddDate(0, 0, 1) + } + return candidate + } + } + interval := time.Duration(job.IntervalSeconds) * time.Second + if interval <= 0 { + interval = 30 * time.Second + } + if initial { + return now + } + return now.Add(interval) +} + +func (p *ControlPlane) runJobOnce(parent context.Context, job ControlledJob, triggerType string, triggerID *int64, configOverride map[string]any) { + cfg, err := p.repo.GetJobForRuntime(parent, job.Key) + if err != nil { + if p.logger != nil { + p.logger.Warn("scheduler job config load failed", zap.String("job_key", job.Key), zap.Error(err)) + } + return + } + isManual := triggerType == opsdomain.SchedulerTriggerManual || triggerType == opsdomain.SchedulerTriggerDryRun + if !cfg.Enabled && !isManual { + return + } + if cfg.ScheduleType == "manual" && !isManual { + return + } + + mergedConfig := cloneMap(cfg.Config) + for key, value := range configOverride { + mergedConfig[key] = value + } + dryRun := triggerType == opsdomain.SchedulerTriggerDryRun || boolFromMap(mergedConfig, "dry_run") + if dryRun { + mergedConfig["dry_run"] = true + } + + lockKey := advisoryLockKey(job.Key) + lockConn, locked := p.tryAdvisoryLock(parent, lockKey) + if !locked { + if p.logger != nil { + p.logger.Info("scheduler job skipped because another instance owns lock", zap.String("job_key", job.Key)) + } + if triggerID != nil { + _ = p.repo.FinishTrigger(parent, *triggerID, nil, opsdomain.SchedulerRunSkipped, controlStringPtr("another scheduler instance is running this job")) + } + return + } + defer p.advisoryUnlock(context.Background(), lockConn, lockKey) + + run, err := p.repo.StartRun(parent, opsrepo.SchedulerRunStart{ + JobKey: job.Key, + TriggerID: triggerID, + SchedulerInstanceID: p.instanceID, + TriggerType: triggerType, + ConfigVersion: cfg.Version, + ConfigSnapshot: schedulerConfigSnapshot(cfg, mergedConfig), + }) + if err != nil { + if p.logger != nil { + p.logger.Warn("scheduler run start failed", zap.String("job_key", job.Key), zap.Error(err)) + } + return + } + + timeout := time.Duration(cfg.TimeoutSeconds) * time.Second + if timeout <= 0 { + timeout = 30 * time.Second + } + ctx, cancel := context.WithTimeout(parent, timeout) + defer cancel() + + stats, runErr := job.Run(ctx, JobRunContext{ + Job: cfg, + TriggerID: triggerID, + Trigger: triggerType, + DryRun: dryRun, + Config: mergedConfig, + }) + status := opsdomain.SchedulerRunSuccess + var errMsg *string + if runErr != nil { + status = opsdomain.SchedulerRunFailed + message := runErr.Error() + errMsg = &message + } + if stats == nil { + stats = map[string]any{} + } + stats["dry_run"] = dryRun + + finished, finishErr := p.repo.FinishRun(parent, opsrepo.SchedulerRunFinish{ + RunID: run.ID, + Status: status, + Stats: stats, + ErrorMessage: errMsg, + }) + if finishErr != nil && p.logger != nil { + p.logger.Warn("scheduler run finish failed", zap.String("job_key", job.Key), zap.Int64("run_id", run.ID), zap.Error(finishErr)) + } + if triggerID != nil { + triggerStatus := status + if finishErr != nil { + triggerStatus = opsdomain.SchedulerRunFailed + } + runID := run.ID + _ = p.repo.FinishTrigger(parent, *triggerID, &runID, triggerStatus, errMsg) + } + if runErr != nil { + if p.logger != nil { + p.logger.Warn("scheduler controlled job failed", zap.String("job_key", job.Key), zap.Error(runErr)) + } + return + } + if p.logger != nil { + duration := int64(0) + if finished != nil && finished.DurationMilliseconds != nil { + duration = *finished.DurationMilliseconds + } + p.logger.Info("scheduler controlled job completed", zap.String("job_key", job.Key), zap.Int64("duration_ms", duration)) + } +} + +func (p *ControlPlane) tryAdvisoryLock(ctx context.Context, key int64) (*pgxpool.Conn, bool) { + conn, err := p.pool.Acquire(ctx) + if err != nil { + if p.logger != nil { + p.logger.Warn("scheduler advisory lock acquire failed", zap.Int64("lock_key", key), zap.Error(err)) + } + return nil, false + } + var ok bool + if err := conn.QueryRow(ctx, `SELECT pg_try_advisory_lock($1)`, key).Scan(&ok); err != nil { + conn.Release() + if p.logger != nil { + p.logger.Warn("scheduler advisory lock failed", zap.Int64("lock_key", key), zap.Error(err)) + } + return nil, false + } + if !ok { + conn.Release() + return nil, false + } + return conn, true +} + +func (p *ControlPlane) advisoryUnlock(ctx context.Context, conn *pgxpool.Conn, key int64) { + if conn == nil { + return + } + defer conn.Release() + _, _ = conn.Exec(ctx, `SELECT pg_advisory_unlock($1)`, key) +} + +func advisoryLockKey(value string) int64 { + hash := fnv.New64a() + _, _ = hash.Write([]byte("scheduler:" + value)) + return int64(hash.Sum64() & 0x7fffffffffffffff) +} + +func schedulerConfigSnapshot(job *opsdomain.SchedulerJob, config map[string]any) map[string]any { + out := map[string]any{ + "enabled": job.Enabled, + "schedule_type": job.ScheduleType, + "interval_seconds": job.IntervalSeconds, + "timezone": job.Timezone, + "timeout_seconds": job.TimeoutSeconds, + "max_concurrency": job.MaxConcurrency, + "config": config, + } + if job.BatchSize != nil { + out["batch_size"] = *job.BatchSize + } + if job.CronExpr != nil { + out["cron_expr"] = *job.CronExpr + } + return out +} + +func cloneMap(in map[string]any) map[string]any { + out := make(map[string]any, len(in)) + for key, value := range in { + out[key] = value + } + return out +} + +func boolFromMap(in map[string]any, key string) bool { + value, ok := in[key] + if !ok { + return false + } + switch typed := value.(type) { + case bool: + return typed + case string: + parsed, _ := strconv.ParseBool(typed) + return parsed + default: + return false + } +} + +func parseRunAtLocal(config map[string]any) string { + value, ok := config["run_at_local"] + if !ok || value == nil { + return "" + } + return fmt.Sprint(value) +} + +func parseHourMinute(value string) (int, int, bool) { + parts := strings.Split(strings.TrimSpace(value), ":") + if len(parts) != 2 { + return 0, 0, false + } + hour, hourErr := strconv.Atoi(parts[0]) + minute, minuteErr := strconv.Atoi(parts[1]) + if hourErr != nil || minuteErr != nil || hour < 0 || hour > 23 || minute < 0 || minute > 59 { + return 0, 0, false + } + return hour, minute, true +} + +func controlStringPtr(value string) *string { + return &value +} + +func AsInt(config map[string]any, key string, fallback int) int { + value, ok := config[key] + if !ok || value == nil { + return fallback + } + switch typed := value.(type) { + case int: + return typed + case int64: + return int(typed) + case float64: + return int(typed) + case string: + parsed, err := strconv.Atoi(typed) + if err == nil { + return parsed + } + } + return fallback +} + +func AsDuration(config map[string]any, key string, fallback time.Duration) time.Duration { + value, ok := config[key] + if !ok || value == nil { + return fallback + } + switch typed := value.(type) { + case time.Duration: + return typed + case string: + parsed, err := time.ParseDuration(typed) + if err == nil { + return parsed + } + case float64: + return time.Duration(typed) * time.Second + } + return fallback +} + +func ErrorFromContext(ctx context.Context) error { + if err := ctx.Err(); err != nil && !errors.Is(err, context.Canceled) { + return fmt.Errorf("scheduler job context finished: %w", err) + } + return nil +} diff --git a/server/internal/scheduler/knowledge_deleted_cleanup_worker.go b/server/internal/scheduler/knowledge_deleted_cleanup_worker.go index 3965053..30eb50c 100644 --- a/server/internal/scheduler/knowledge_deleted_cleanup_worker.go +++ b/server/internal/scheduler/knowledge_deleted_cleanup_worker.go @@ -45,3 +45,14 @@ func (w *KnowledgeDeletedCleanupWorker) run(ctx context.Context) { } } } + +func (w *KnowledgeDeletedCleanupWorker) RunOnce(ctx context.Context) (map[string]any, error) { + if w == nil || w.service == nil { + return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil + } + startedAt := time.Now() + w.service.RunDeletedCleanupSweep() + return map[string]any{ + "duration_ms": time.Since(startedAt).Milliseconds(), + }, nil +} diff --git a/server/internal/scheduler/monitoring_lease_recovery_worker.go b/server/internal/scheduler/monitoring_lease_recovery_worker.go index 342bfa9..eb9636e 100644 --- a/server/internal/scheduler/monitoring_lease_recovery_worker.go +++ b/server/internal/scheduler/monitoring_lease_recovery_worker.go @@ -90,3 +90,30 @@ func (w *MonitoringLeaseRecoveryWorker) runOnce(parent context.Context) { w.logger.Info("monitoring lease recovery completed", zap.Int64("expired_task_count", expiredCount)) } } + +func (w *MonitoringLeaseRecoveryWorker) RunOnce(parent context.Context) (map[string]any, error) { + if w == nil || w.monitoringPool == nil { + return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil + } + startedAt := time.Now() + ctx, cancel := context.WithTimeout(parent, w.timeout) + defer cancel() + + tx, err := w.monitoringPool.Begin(ctx) + if err != nil { + return map[string]any{"stage": "begin"}, err + } + defer tx.Rollback(ctx) + + expiredCount, err := tenantapp.ExpireMonitoringLeasedTasks(ctx, tx, nil, time.Now().UTC()) + if err != nil { + return map[string]any{"stage": "expire"}, err + } + if err := tx.Commit(ctx); err != nil { + return map[string]any{"stage": "commit"}, err + } + return map[string]any{ + "expired_task_count": expiredCount, + "duration_ms": time.Since(startedAt).Milliseconds(), + }, nil +} diff --git a/server/internal/scheduler/monitoring_received_inspection_worker.go b/server/internal/scheduler/monitoring_received_inspection_worker.go index 18481fb..4121473 100644 --- a/server/internal/scheduler/monitoring_received_inspection_worker.go +++ b/server/internal/scheduler/monitoring_received_inspection_worker.go @@ -116,3 +116,32 @@ func (w *MonitoringReceivedInspectionWorker) runOnce(parent context.Context) { ) } } + +func (w *MonitoringReceivedInspectionWorker) RunOnce(parent context.Context) (map[string]any, error) { + if w == nil || w.monitoringPool == nil { + return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil + } + startedAt := time.Now() + ctx, cancel := context.WithTimeout(parent, w.timeout) + defer cancel() + + tx, err := w.monitoringPool.Begin(ctx) + if err != nil { + return map[string]any{"stage": "begin"}, err + } + defer tx.Rollback(ctx) + + now := time.Now().UTC() + items, err := tenantapp.MarkMonitoringStaleReceivedTasks(ctx, tx, now, w.threshold, w.limit) + if err != nil { + return map[string]any{"stage": "inspect"}, err + } + if err := tx.Commit(ctx); err != nil { + return map[string]any{"stage": "commit"}, err + } + return map[string]any{ + "overdue_task_count": len(items), + "threshold_seconds": int64(w.threshold.Seconds()), + "duration_ms": time.Since(startedAt).Milliseconds(), + }, nil +} diff --git a/server/internal/scheduler/monitoring_result_recovery_worker.go b/server/internal/scheduler/monitoring_result_recovery_worker.go index 8f03efe..4e0b42b 100644 --- a/server/internal/scheduler/monitoring_result_recovery_worker.go +++ b/server/internal/scheduler/monitoring_result_recovery_worker.go @@ -118,6 +118,41 @@ func (w *MonitoringResultRecoveryWorker) runOnce(parent context.Context) { } } +func (w *MonitoringResultRecoveryWorker) RunOnce(parent context.Context) (map[string]any, error) { + if w == nil || w.monitoringPool == nil || w.service == nil { + return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil + } + startedAt := time.Now() + ctx, cancel := context.WithTimeout(parent, w.timeout) + defer cancel() + + tx, err := w.monitoringPool.Begin(ctx) + if err != nil { + return map[string]any{"stage": "begin"}, err + } + defer tx.Rollback(ctx) + + items, err := loadMonitoringRecoverableTaskResults(ctx, tx, w.limit) + if err != nil { + return map[string]any{"stage": "load"}, err + } + if err := tx.Commit(ctx); err != nil { + return map[string]any{"stage": "commit"}, err + } + + republishedCount := 0 + for _, item := range items { + if w.tryRepublish(parent, item) { + republishedCount++ + } + } + return map[string]any{ + "recoverable_count": len(items), + "republished_count": republishedCount, + "duration_ms": time.Since(startedAt).Milliseconds(), + }, nil +} + func (w *MonitoringResultRecoveryWorker) tryRepublish(parent context.Context, item monitoringRecoverableTaskResult) bool { if err := tenantapp.ValidateMonitoringTaskResultEnvelope(item.RequestPayloadJSON); err != nil { w.service.PatchTaskResultQueueMeta(item.ID, item.CallbackReceivedAt, tenantapp.MonitoringTaskResultQueueMetaPatch{ diff --git a/server/internal/scheduler/monitoring_retention_worker.go b/server/internal/scheduler/monitoring_retention_worker.go new file mode 100644 index 0000000..2b5c1ba --- /dev/null +++ b/server/internal/scheduler/monitoring_retention_worker.go @@ -0,0 +1,203 @@ +package scheduler + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" +) + +const ( + defaultMonitoringRetentionDays = 30 + defaultMonitoringRetentionBatch = 5000 + defaultMonitoringRetentionBatches = 200 +) + +type MonitoringRetentionWorker struct { + monitoringPool *pgxpool.Pool + logger *zap.Logger +} + +type retentionTarget struct { + Name string + DeleteSQL string + CountSQL string +} + +func NewMonitoringRetentionWorker(monitoringPool *pgxpool.Pool, logger *zap.Logger) *MonitoringRetentionWorker { + return &MonitoringRetentionWorker{monitoringPool: monitoringPool, logger: logger} +} + +func (w *MonitoringRetentionWorker) RunOnce(ctx context.Context, run JobRunContext) (map[string]any, error) { + if w == nil || w.monitoringPool == nil { + return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil + } + startedAt := time.Now() + retentionDays := AsInt(run.Config, "retention_days", defaultMonitoringRetentionDays) + if retentionDays < defaultMonitoringRetentionDays { + retentionDays = defaultMonitoringRetentionDays + } + batchSize := AsInt(run.Config, "batch_size", defaultMonitoringRetentionBatch) + if run.Job != nil && run.Job.BatchSize != nil { + batchSize = *run.Job.BatchSize + } + if batchSize <= 0 { + batchSize = defaultMonitoringRetentionBatch + } + maxBatches := AsInt(run.Config, "max_batches_per_run", defaultMonitoringRetentionBatches) + if maxBatches <= 0 { + maxBatches = defaultMonitoringRetentionBatches + } + statementTimeout := AsDuration(run.Config, "statement_timeout", 5*time.Second) + cutoffDate := monitoringRetentionCutoffDate(time.Now(), retentionDays) + dryRun := run.DryRun + + stats := map[string]any{ + "retention_days": retentionDays, + "cutoff_date": cutoffDate, + "batch_size": batchSize, + "max_batches_per_run": maxBatches, + "dry_run": dryRun, + } + + conn, err := w.monitoringPool.Acquire(ctx) + if err != nil { + return stats, fmt.Errorf("acquire monitoring connection: %w", err) + } + defer conn.Release() + + if statementTimeout > 0 { + if _, err := conn.Exec(ctx, `SET statement_timeout = $1`, int(statementTimeout.Milliseconds())); err != nil { + return stats, fmt.Errorf("set statement timeout: %w", err) + } + defer conn.Exec(context.Background(), `RESET statement_timeout`) + } + + totalDeleted := int64(0) + totalCandidates := int64(0) + perTable := map[string]any{} + for _, target := range retentionTargets() { + candidateCount, err := w.countTarget(ctx, conn, target, cutoffDate) + if err != nil { + return stats, fmt.Errorf("count %s: %w", target.Name, err) + } + totalCandidates += candidateCount + tableStats := map[string]any{ + "candidate_count": candidateCount, + } + if !dryRun && candidateCount > 0 { + deleted, batches, err := w.deleteTarget(ctx, conn, target, cutoffDate, batchSize, maxBatches) + if err != nil { + return stats, fmt.Errorf("delete %s: %w", target.Name, err) + } + tableStats["deleted_count"] = deleted + tableStats["batch_count"] = batches + totalDeleted += deleted + } + perTable[target.Name] = tableStats + } + + stats["candidate_count"] = totalCandidates + stats["deleted_count"] = totalDeleted + stats["tables"] = perTable + stats["duration_ms"] = time.Since(startedAt).Milliseconds() + return stats, nil +} + +func monitoringRetentionCutoffDate(now time.Time, retentionDays int) string { + loc, err := time.LoadLocation("Asia/Shanghai") + if err != nil { + loc = time.FixedZone("UTC+8", 8*60*60) + } + localNow := now.In(loc) + today := time.Date(localNow.Year(), localNow.Month(), localNow.Day(), 0, 0, 0, 0, loc) + cutoff := today.AddDate(0, 0, -(retentionDays - 1)) + return cutoff.Format("2006-01-02") +} + +func (w *MonitoringRetentionWorker) countTarget(ctx context.Context, conn *pgxpool.Conn, target retentionTarget, cutoffDate string) (int64, error) { + var count int64 + if err := conn.QueryRow(ctx, target.CountSQL, cutoffDate).Scan(&count); err != nil { + return 0, err + } + return count, nil +} + +func (w *MonitoringRetentionWorker) deleteTarget(ctx context.Context, conn *pgxpool.Conn, target retentionTarget, cutoffDate string, batchSize, maxBatches int) (int64, int, error) { + total := int64(0) + batches := 0 + for batches < maxBatches { + tag, err := conn.Exec(ctx, target.DeleteSQL, cutoffDate, batchSize) + if err != nil { + return total, batches, err + } + affected := tag.RowsAffected() + if affected == 0 { + break + } + total += affected + batches++ + if affected < int64(batchSize) { + break + } + } + return total, batches, nil +} + +func retentionTargets() []retentionTarget { + return []retentionTarget{ + { + Name: "monitoring_collect_dispatch_outbox", + CountSQL: `SELECT COUNT(*) FROM monitoring_collect_dispatch_outbox WHERE created_at < $1::date`, + DeleteSQL: batchDeleteSQL("monitoring_collect_dispatch_outbox", "id", "created_at < $1::date"), + }, + { + Name: "monitoring_collect_requests", + CountSQL: `SELECT COUNT(*) FROM monitoring_collect_requests WHERE created_at < $1::date`, + DeleteSQL: batchDeleteSQL("monitoring_collect_requests", "id", "created_at < $1::date"), + }, + { + Name: "monitoring_brand_platform_daily", + CountSQL: `SELECT COUNT(*) FROM monitoring_brand_platform_daily WHERE business_date < $1::date`, + DeleteSQL: batchDeleteSQL("monitoring_brand_platform_daily", "id", "business_date < $1::date"), + }, + { + Name: "monitoring_brand_daily", + CountSQL: `SELECT COUNT(*) FROM monitoring_brand_daily WHERE business_date < $1::date`, + DeleteSQL: batchDeleteSQL("monitoring_brand_daily", "id", "business_date < $1::date"), + }, + { + Name: "monitoring_platform_access_snapshots", + CountSQL: `SELECT COUNT(*) FROM monitoring_platform_access_snapshots WHERE business_date < $1::date`, + DeleteSQL: batchDeleteSQL("monitoring_platform_access_snapshots", "id", "business_date < $1::date"), + }, + { + Name: "monitoring_collect_tasks", + CountSQL: `SELECT COUNT(*) FROM monitoring_collect_tasks WHERE business_date < $1::date`, + DeleteSQL: batchDeleteSQL("monitoring_collect_tasks", "id", "business_date < $1::date"), + }, + { + Name: "question_monitor_runs", + CountSQL: `SELECT COUNT(*) FROM question_monitor_runs WHERE business_date < $1::date`, + DeleteSQL: batchDeleteSQL("question_monitor_runs", "id", "business_date < $1::date"), + }, + } +} + +func batchDeleteSQL(table, idColumn, predicate string) string { + parts := []string{ + "WITH doomed AS (", + "SELECT " + idColumn, + "FROM " + table, + "WHERE " + predicate, + "ORDER BY " + idColumn + " ASC", + "LIMIT $2", + ")", + "DELETE FROM " + table, + "WHERE " + idColumn + " IN (SELECT " + idColumn + " FROM doomed)", + } + return strings.Join(parts, "\n") +} diff --git a/server/internal/scheduler/schedule_dispatch_worker.go b/server/internal/scheduler/schedule_dispatch_worker.go index 21d2f34..a790f1a 100644 --- a/server/internal/scheduler/schedule_dispatch_worker.go +++ b/server/internal/scheduler/schedule_dispatch_worker.go @@ -173,6 +173,57 @@ func (w *ScheduleDispatchWorker) runOnce(parent context.Context) { w.dispatchBatch(parent, tasks) } +func (w *ScheduleDispatchWorker) RunOnce(ctx context.Context) (map[string]any, error) { + if w == nil || w.pool == nil || w.promptRuleService == nil { + return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil + } + startedAt := time.Now() + + stageCtx, cancel := w.stageContext(ctx) + hydratedCount, err := w.hydrateMissingNextRuns(stageCtx) + cancel() + if err != nil { + return map[string]any{ + "stage": "hydrate_missing_next_runs", + }, err + } + + stageCtx, cancel = w.stageContext(ctx) + paused, stats, err := w.shouldPauseDispatch(stageCtx) + cancel() + if err != nil { + return map[string]any{ + "stage": "queue_backpressure", + "hydrated_count": hydratedCount, + }, err + } + if paused { + return map[string]any{ + "hydrated_count": hydratedCount, + "paused": true, + "queue_depth": stats.Messages, + "queue_consumers": stats.Consumers, + "duration_ms": time.Since(startedAt).Milliseconds(), + }, nil + } + + stageCtx, cancel = w.stageContext(ctx) + tasks, err := w.claimDueSchedules(stageCtx) + cancel() + if err != nil { + return map[string]any{ + "stage": "claim_due_schedules", + "hydrated_count": hydratedCount, + }, err + } + w.dispatchBatch(ctx, tasks) + return map[string]any{ + "hydrated_count": hydratedCount, + "claimed_count": len(tasks), + "duration_ms": time.Since(startedAt).Milliseconds(), + }, nil +} + func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context) (int, error) { runtimeCfg := w.runtimeConfig() tx, err := w.pool.Begin(ctx) diff --git a/server/internal/tenant/app/monitoring_daily_task_worker.go b/server/internal/tenant/app/monitoring_daily_task_worker.go index 8352026..4882142 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker.go @@ -174,6 +174,42 @@ func (w *MonitoringDailyTaskWorker) runOnce(parent context.Context) { ) } +func (w *MonitoringDailyTaskWorker) RunOnce(parent context.Context) (map[string]any, error) { + if w == nil || w.service == nil || w.service.businessPool == nil || w.service.monitoringPool == nil { + return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil + } + ctx, cancel := context.WithTimeout(parent, w.timeout) + defer cancel() + + startedAt := time.Now() + summary, err := w.service.GenerateDailyMonitoringTasks(ctx, w.projectionService, time.Now().UTC()) + duration := time.Since(startedAt) + recordMonitoringDailyTaskMetrics(summary, duration, err) + stats := map[string]any{ + "duration_ms": duration.Milliseconds(), + } + if summary != nil { + stats["business_date"] = summary.BusinessDate + stats["tenant_count"] = summary.TenantCount + stats["brand_count"] = summary.BrandCount + stats["failed_plan_count"] = summary.FailedPlanCount + stats["failed_brand_count"] = summary.FailedBrandCount + stats["planned_task_count"] = summary.PlannedTaskCount + stats["due_task_count"] = summary.DueTaskCount + stats["created_task_count"] = summary.CreatedTaskCount + stats["desktop_task_count"] = summary.DesktopTaskCount + stats["deferred_task_count"] = summary.DeferredTaskCount + stats["skipped_plan_count"] = summary.SkippedPlanCount + } + if err != nil { + if w.logger != nil { + w.logger.Warn("monitoring daily task generation failed", MonitoringWorkerErrorFields(err)...) + } + return stats, err + } + return stats, nil +} + func (s *MonitoringService) GenerateDailyMonitoringTasks( ctx context.Context, projectionService *MonitoringCallbackService, diff --git a/server/internal/worker/generate/article_generation_state_check_worker.go b/server/internal/worker/generate/article_generation_state_check_worker.go index 9c9341d..282ae23 100644 --- a/server/internal/worker/generate/article_generation_state_check_worker.go +++ b/server/internal/worker/generate/article_generation_state_check_worker.go @@ -92,6 +92,34 @@ func (w *GenerationTaskStateCheckWorker) runOnce(parent context.Context) { }, zap.Int("anomaly_count", count)) } +func (w *GenerationTaskStateCheckWorker) RunOnce(parent context.Context) (map[string]any, error) { + if w == nil || w.pool == nil { + return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil + } + cfg := w.runtimeConfig() + startedAt := time.Now() + count, err := CheckGenerationTaskStateConsistency(parent, w.pool, w.logger, w.cache, cfg) + duration := time.Since(startedAt) + stats := map[string]any{ + "anomaly_count": count, + "duration_ms": duration.Milliseconds(), + "batch_size": cfg.BatchSize, + "lookback_sec": int64(cfg.Lookback.Seconds()), + } + if err != nil { + logCtx := tenantapp.GenerationTaskLogContext{ + Stage: "state_check", + Result: tenantapp.GenerationTaskResultFailure, + Severity: "warning", + Duration: duration, + } + tenantapp.RecordGenerationTaskEvent(tenantapp.GenerationTaskEventStateCheckFailure, logCtx) + tenantapp.LogGenerationTaskWarn(w.logger, "article generation state consistency check failed", logCtx, zap.Error(err)) + return stats, err + } + return stats, nil +} + func (w *GenerationTaskStateCheckWorker) runtimeConfig() generationTaskStateCheckConfig { if w != nil && w.provider != nil { if cfg := w.provider.Current(); cfg != nil {