feat: add ops scheduler control center

2026-05-20 10:46:49 +08:00
parent 98f73c5bea
commit e82ae56236
19 changed files with 2766 additions and 20 deletions
@@ -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
+}
@@ -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
+}
@@ -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
+}
@@ -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
+}
@@ -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{
@@ -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")
+}
@@ -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)