geo/server/internal/worker/generate/article_generation_state_check.go

package generate

import (
	"context"
	"database/sql"
	"time"

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

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

const (
	generationTaskAnomalyArticleFailedTaskActive     = "article_failed_task_active"
	generationTaskAnomalyArticleCompletedTaskActive  = "article_completed_task_active"
	generationTaskAnomalyCompletedTaskArticleMissing = "completed_task_article_not_completed"
	generationTaskAnomalyCompletedTaskMissingVersion = "completed_task_missing_version"
	generationTaskAnomalyExpiredRunningLease         = "expired_running_lease"
)

type generationTaskStateAnomaly struct {
	TaskID        int64
	TenantID      int64
	OperatorID    int64
	ArticleID     int64
	ReservationID int64
	TaskType      string
	TaskStatus    string
	ArticleStatus string
	AttemptCount  int
	LeaseOwner    string
	AnomalyType   string
	Severity      string
}

func CheckGenerationTaskStateConsistency(ctx context.Context, pool generationTaskStateDB, logger *zap.Logger, cfg generationTaskStateCheckConfig) (int, error) {
	if pool == nil || cfg.BatchSize <= 0 {
		return 0, nil
	}
	if cfg.Timeout > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, cfg.Timeout)
		defer cancel()
	}
	lookback := time.Now().UTC().Add(-cfg.Lookback)

	rows, err := pool.Query(ctx, `
		WITH candidates AS (
			SELECT
				gt.id,
				gt.tenant_id,
				gt.operator_id,
				gt.article_id,
				gt.quota_reservation_id,
				gt.task_type,
				gt.status AS task_status,
				a.generate_status AS article_status,
				gt.attempt_count,
				gt.lease_owner,
				CASE
					WHEN a.generate_status = 'failed' AND gt.status IN ('queued', 'running') THEN 'article_failed_task_active'
					WHEN a.generate_status = 'completed' AND gt.status IN ('queued', 'running') THEN 'article_completed_task_active'
					WHEN gt.status = 'completed' AND COALESCE(a.generate_status, '') <> 'completed' THEN 'completed_task_article_not_completed'
					WHEN gt.status = 'completed' AND (a.current_version_id IS NULL OR av.id IS NULL) THEN 'completed_task_missing_version'
					WHEN gt.status = 'running' AND (gt.lease_expires_at IS NULL OR gt.lease_expires_at < NOW()) THEN 'expired_running_lease'
					ELSE ''
				END AS anomaly_type
			FROM generation_tasks gt
			LEFT JOIN articles a
			  ON a.id = gt.article_id
			 AND a.tenant_id = gt.tenant_id
			 AND a.deleted_at IS NULL
			LEFT JOIN article_versions av
			  ON av.id = a.current_version_id
			WHERE gt.article_id IS NOT NULL
			  AND gt.created_at >= $1
			  AND (
				gt.status IN ('queued', 'running')
				OR gt.updated_at >= $1
			  )
			  AND (
				(a.generate_status IN ('failed', 'completed') AND gt.status IN ('queued', 'running'))
				OR (gt.status = 'completed' AND COALESCE(a.generate_status, '') <> 'completed')
				OR (gt.status = 'completed' AND (a.current_version_id IS NULL OR av.id IS NULL))
				OR (gt.status = 'running' AND (gt.lease_expires_at IS NULL OR gt.lease_expires_at < NOW()))
			  )
			ORDER BY gt.updated_at DESC, gt.id DESC
			LIMIT $2
		)
		SELECT id, tenant_id, operator_id, article_id, quota_reservation_id, task_type, task_status, article_status, attempt_count, lease_owner, anomaly_type
		FROM candidates
		WHERE anomaly_type <> ''
	`, lookback, cfg.BatchSize)
	if err != nil {
		return 0, err
	}
	defer rows.Close()

	count := 0
	for rows.Next() {
		item, err := scanGenerationTaskStateAnomaly(rows)
		if err != nil {
			return count, err
		}
		if item.AnomalyType == "" {
			continue
		}
		count++
		logCtx := tenantapp.GenerationTaskLogContext{
			TaskID:        item.TaskID,
			TenantID:      item.TenantID,
			OperatorID:    item.OperatorID,
			ArticleID:     item.ArticleID,
			ReservationID: item.ReservationID,
			TaskType:      item.TaskType,
			Stage:         "state_check",
			AttemptCount:  item.AttemptCount,
			LeaseOwner:    item.LeaseOwner,
			StatusBefore:  item.TaskStatus,
			StatusAfter:   item.ArticleStatus,
			Result:        tenantapp.GenerationTaskResultFailure,
			AnomalyType:   item.AnomalyType,
			Severity:      item.Severity,
		}
		tenantapp.RecordGenerationTaskEvent(tenantapp.GenerationTaskEventStateAnomaly, logCtx)
		tenantapp.LogGenerationTaskError(logger, "article generation state consistency anomaly detected", logCtx,
			zap.String("task_status", item.TaskStatus),
			zap.String("article_status", item.ArticleStatus),
		)
	}
	if err := rows.Err(); err != nil {
		return count, err
	}
	return count, nil
}

type generationTaskStateAnomalyRow interface {
	Scan(dest ...any) error
}

type generationTaskStateDB interface {
	Query(context.Context, string, ...any) (pgx.Rows, error)
}

func scanGenerationTaskStateAnomaly(row generationTaskStateAnomalyRow) (generationTaskStateAnomaly, error) {
	var (
		item          generationTaskStateAnomaly
		operatorID    sql.NullInt64
		articleID     sql.NullInt64
		reservationID sql.NullInt64
		articleStatus sql.NullString
		leaseOwner    sql.NullString
	)
	if err := row.Scan(
		&item.TaskID,
		&item.TenantID,
		&operatorID,
		&articleID,
		&reservationID,
		&item.TaskType,
		&item.TaskStatus,
		&articleStatus,
		&item.AttemptCount,
		&leaseOwner,
		&item.AnomalyType,
	); err != nil {
		return generationTaskStateAnomaly{}, err
	}
	item.OperatorID = operatorID.Int64
	item.ArticleID = articleID.Int64
	item.ReservationID = reservationID.Int64
	item.ArticleStatus = articleStatus.String
	item.LeaseOwner = leaseOwner.String
	item.Severity = generationTaskAnomalySeverity(item.AnomalyType)
	return item, nil
}

func generationTaskAnomalySeverity(anomalyType string) string {
	switch anomalyType {
	case generationTaskAnomalyCompletedTaskArticleMissing,
		generationTaskAnomalyCompletedTaskMissingVersion,
		generationTaskAnomalyArticleFailedTaskActive:
		return "critical"
	case generationTaskAnomalyArticleCompletedTaskActive,
		generationTaskAnomalyExpiredRunningLease:
		return "warning"
	default:
		return "warning"
	}
}