sub2api-cn-relay-manager/internal/batch/validation.go

package batch

import (
	"context"
	"fmt"
	"strings"

	"sub2api-cn-relay-manager/internal/host/sub2api"
	"sub2api-cn-relay-manager/internal/store/sqlite"
)

type ValidationItemStore interface {
	Upsert(ctx context.Context, item sqlite.ImportRunItem) error
}

type ValidationRunStore interface {
	GetByRunID(ctx context.Context, runID string) (sqlite.ImportRun, error)
	Update(ctx context.Context, run sqlite.ImportRun) error
}

type ValidationService struct {
	ItemStore ValidationItemStore
	RunStore  ValidationRunStore
	Validator func(ctx context.Context, item sqlite.ImportRunItem) (sub2api.GatewayCompletionResult, error)
}

func (s ValidationService) ValidateItem(ctx context.Context, item sqlite.ImportRunItem) error {
	if s.ItemStore == nil {
		return fmt.Errorf("item store is required")
	}
	if s.RunStore == nil {
		return fmt.Errorf("run store is required")
	}
	if s.Validator == nil {
		return fmt.Errorf("validator is required")
	}
	if strings.TrimSpace(item.CurrentStage) != string(ItemStageValidate) {
		return fmt.Errorf("item %s is not ready for validation", strings.TrimSpace(item.ItemID))
	}

	completion, err := s.Validator(ctx, item)
	if err != nil {
		return err
	}

	item.CurrentStage = string(ItemStageDone)
	item.AccessStatus = string(resolveValidationAccessStatus(item.ConfirmationStatus, completion))
	if item.AccessStatus == string(AccessStatusDegraded) {
		item.AdvisoryMessagesJSON = appendAdvisoryJSON(item.AdvisoryMessagesJSON, "gateway_warmup_retry_succeeded")
	}
	if !completion.OK {
		item.LastErrorStage = string(ItemStageValidate)
		item.LastError = strings.TrimSpace(completion.BodyPreview)
	}

	if err := s.ItemStore.Upsert(ctx, item); err != nil {
		return err
	}

	run, err := s.RunStore.GetByRunID(ctx, item.RunID)
	if err != nil {
		return err
	}

	run.CompletedItems++
	switch item.AccessStatus {
	case string(AccessStatusActive):
		run.ActiveItems++
	case string(AccessStatusDegraded):
		run.DegradedItems++
		run.WarningItems++
	case string(AccessStatusBroken):
		run.BrokenItems++
	}
	run.State = deriveRunState(run)

	return s.RunStore.Update(ctx, run)
}

func resolveValidationAccessStatus(confirmationStatus string, completion sub2api.GatewayCompletionResult) AccessStatus {
	switch strings.TrimSpace(confirmationStatus) {
	case string(ConfirmationFailed):
		return AccessStatusBroken
	case string(ConfirmationConfirmed), string(ConfirmationAdvisory):
		if completion.OK && completion.StatusCode >= 200 && completion.StatusCode < 300 {
			return AccessStatusActive
		}
		if isTransientValidationFailure(completion) {
			return AccessStatusDegraded
		}
		return AccessStatusBroken
	default:
		return AccessStatusBroken
	}
}

func isTransientValidationFailure(result sub2api.GatewayCompletionResult) bool {
	if result.OK {
		return false
	}
	if result.StatusCode != 0 && result.StatusCode != 429 && result.StatusCode != 502 && result.StatusCode != 503 && result.StatusCode != 504 {
		return false
	}

	body := strings.ToLower(strings.TrimSpace(result.BodyPreview))
	return strings.Contains(body, "service temporarily unavailable") ||
		strings.Contains(body, "no available accounts") ||
		strings.Contains(body, "temporar") ||
		strings.Contains(body, "try again")
}

func deriveRunState(run sqlite.ImportRun) string {
	if run.TotalItems > 0 && run.CompletedItems >= run.TotalItems {
		switch {
		case run.BrokenItems > 0:
			return string(RunStateFailed)
		case run.WarningItems > 0 || run.DegradedItems > 0:
			return string(RunStateCompletedWithWarnings)
		default:
			return string(RunStateCompleted)
		}
	}
	return firstNonEmptyRunState(run.State, string(RunStateRunning))
}

func firstNonEmptyRunState(values ...string) string {
	for _, value := range values {
		if trimmed := strings.TrimSpace(value); trimmed != "" {
			return trimmed
		}
	}
	return string(RunStateRunning)
}