geo/server/internal/scheduler/monitoring_received_inspection_worker.go

package scheduler

import (
	"context"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
	"go.uber.org/zap"

	tenantapp "github.com/geo-platform/tenant-api/internal/tenant/app"
)

const (
	defaultMonitoringReceivedInspectionInterval = 5 * time.Minute
	defaultMonitoringReceivedInspectionTimeout  = 30 * time.Second
	defaultMonitoringReceivedAlertThreshold     = 15 * time.Minute
	defaultMonitoringReceivedInspectLimit       = 100
)

type MonitoringReceivedInspectionWorker struct {
	monitoringPool *pgxpool.Pool
	logger         *zap.Logger
	interval       time.Duration
	timeout        time.Duration
	threshold      time.Duration
	limit          int
}

func NewMonitoringReceivedInspectionWorker(monitoringPool *pgxpool.Pool, logger *zap.Logger) *MonitoringReceivedInspectionWorker {
	return &MonitoringReceivedInspectionWorker{
		monitoringPool: monitoringPool,
		logger:         logger,
		interval:       defaultMonitoringReceivedInspectionInterval,
		timeout:        defaultMonitoringReceivedInspectionTimeout,
		threshold:      defaultMonitoringReceivedAlertThreshold,
		limit:          defaultMonitoringReceivedInspectLimit,
	}
}

func (w *MonitoringReceivedInspectionWorker) Start(ctx context.Context) {
	go w.Run(ctx)
}

func (w *MonitoringReceivedInspectionWorker) Run(ctx context.Context) {
	if w == nil || w.monitoringPool == nil {
		return
	}
	w.run(ctx)
}

func (w *MonitoringReceivedInspectionWorker) run(ctx context.Context) {
	w.runOnce(context.Background())

	ticker := time.NewTicker(w.interval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			w.runOnce(context.Background())
		}
	}
}

func (w *MonitoringReceivedInspectionWorker) runOnce(parent context.Context) {
	ctx, cancel := context.WithTimeout(parent, w.timeout)
	defer cancel()

	tx, err := w.monitoringPool.Begin(ctx)
	if err != nil {
		if w.logger != nil {
			w.logger.Warn("monitoring received watchdog begin failed", zap.Error(err))
		}
		return
	}
	defer tx.Rollback(ctx)

	now := time.Now().UTC()
	items, err := tenantapp.MarkMonitoringStaleReceivedTasks(ctx, tx, now, w.threshold, w.limit)
	if err != nil {
		if w.logger != nil {
			w.logger.Warn("monitoring received watchdog failed", tenantapp.MonitoringWorkerErrorFields(err)...)
		}
		return
	}

	if err := tx.Commit(ctx); err != nil {
		if w.logger != nil {
			w.logger.Warn("monitoring received watchdog commit failed", zap.Error(err))
		}
		return
	}

	if len(items) == 0 || w.logger == nil {
		return
	}

	w.logger.Warn("monitoring received watchdog found overdue tasks",
		zap.Int("task_count", len(items)),
		zap.Duration("threshold", w.threshold),
	)

	for _, item := range items {
		w.logger.Warn("monitoring task stuck in received state",
			zap.Int64("task_id", item.ID),
			zap.Int64("tenant_id", item.TenantID),
			zap.Int64("brand_id", item.BrandID),
			zap.Int64("question_id", item.QuestionID),
			zap.String("collector_type", item.CollectorType),
			zap.String("ai_platform_id", item.AIPlatformID),
			zap.String("business_date", item.BusinessDate.Format("2006-01-02")),
			zap.Time("callback_received_at", item.CallbackReceivedAt.UTC()),
			zap.Duration("received_lag", now.Sub(item.CallbackReceivedAt.UTC())),
		)
	}
}

func (w *MonitoringReceivedInspectionWorker) RunOnce(parent 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()
	timeout := w.timeout
	if run.Job != nil && run.Job.TimeoutSeconds > 0 {
		timeout = time.Duration(run.Job.TimeoutSeconds) * time.Second
	}
	limit := w.limit
	if run.Job != nil && run.Job.BatchSize != nil && *run.Job.BatchSize > 0 {
		limit = *run.Job.BatchSize
	}
	if value := AsInt(run.Config, "batch_size", 0); value > 0 && (run.Job == nil || run.Job.BatchSize == nil) {
		limit = value
	}
	threshold := w.threshold
	if value := AsDuration(run.Config, "threshold", 0); value > 0 {
		threshold = value
	}
	ctx, cancel := context.WithTimeout(parent, 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, threshold, 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(threshold.Seconds()),
		"duration_ms":        time.Since(startedAt).Milliseconds(),
		"batch_size":         limit,
	}, nil
}