package service

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"math/rand"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/credentials/stscreds"
	"github.com/aws/aws-sdk-go-v2/service/secretsmanager"
	"github.com/aws/aws-sdk-go-v2/service/sts"

	coremodel "security-center-backend/internal/model"
	"security-center-backend/internal/modules/caasm/collector"
	aws_collector "security-center-backend/internal/modules/caasm/collector/aws"
	caasmmodel "security-center-backend/internal/modules/caasm/model"
	caasmqueue "security-center-backend/internal/modules/caasm/queue"
	"security-center-backend/internal/modules/caasm/repository"
	"security-center-backend/internal/pkg/logger"
	corerepo "security-center-backend/internal/repository"
)

// BackoffError represents an error when a sync is rate-limited due to recent failures.
type BackoffError struct {
	AccountID uint64
	RetryAt   time.Time
}

func (e *BackoffError) Error() string {
	return fmt.Sprintf("backoff active for account %d; retry after %s", e.AccountID, e.RetryAt.Format(time.RFC3339))
}

type AssetService struct {
	assets repository.AssetRepository
	users  corerepo.UserRepository
}

type AssetListResult struct {
	Items []caasmmodel.Asset
	Total int64
}

type IPAssetListResult struct {
	Items []caasmmodel.IPAsset
	Total int64
}

type AssetDetailResult struct {
	Asset         caasmmodel.Asset
	IPAddresses   []caasmmodel.IPAsset
	DomainAssets  []caasmmodel.DomainAsset
	Relationships []caasmmodel.AssetRelationship
}

type RelationshipGraph struct {
	Nodes []GraphNode `json:"nodes"`
	Edges []GraphEdge `json:"edges"`
}

// relationshipGraphMaxEdges 限制关系图返回的边数，避免全量 asset_relationships
// 把响应和前端力导图拖垮；超出部分按 id 升序截断（见 repository.RelationshipFilter.Limit）。
const relationshipGraphMaxEdges = 1000

type GraphNode struct {
	ID    string `json:"id"`
	Label string `json:"label"`
	Type  string `json:"type"`
}

type GraphEdge struct {
	Source string `json:"source"`
	Target string `json:"target"`
	Label  string `json:"label"`
}

func NewAssetService(repo repository.AssetRepository, userRepo corerepo.UserRepository) *AssetService {
	return &AssetService{assets: repo, users: userRepo}
}

func (s *AssetService) List(ctx context.Context, filter repository.AssetFilter, page, pageSize int) (*AssetListResult, error) {
	if s.assets == nil {
		return &AssetListResult{}, nil
	}
	items, total, err := s.assets.List(ctx, filter, page, pageSize)
	if err != nil {
		return nil, err
	}
	return &AssetListResult{Items: items, Total: total}, nil
}

func (s *AssetService) ListIPAssets(ctx context.Context, filter repository.IPAssetFilter, page, pageSize int) (*IPAssetListResult, error) {
	if s.assets == nil {
		return &IPAssetListResult{}, nil
	}
	items, total, err := s.assets.ListIPAssets(ctx, filter, page, pageSize)
	if err != nil {
		return nil, err
	}
	return &IPAssetListResult{Items: items, Total: total}, nil
}

func (s *AssetService) Detail(ctx context.Context, id uint64) (*AssetDetailResult, error) {
	if s.assets == nil {
		return nil, errors.New("asset repository is not configured")
	}
	detail, err := s.assets.Detail(ctx, id)
	if err != nil {
		return nil, err
	}
	return &AssetDetailResult{
		Asset:         detail.Asset,
		IPAddresses:   detail.IPAddresses,
		DomainAssets:  detail.DomainAssets,
		Relationships: detail.Relationships,
	}, nil
}

func (s *AssetService) GetRelationshipGraph(ctx context.Context) (*RelationshipGraph, error) {
	if s.assets == nil {
		return nil, errors.New("asset repository is not configured")
	}
	graph := &RelationshipGraph{Nodes: []GraphNode{}, Edges: []GraphEdge{}}

	rels, err := s.assets.ListRelationships(ctx, repository.RelationshipFilter{Limit: relationshipGraphMaxEdges})
	if err != nil {
		return nil, err
	}
	if len(rels) == 0 {
		return graph, nil
	}

	seen := make(map[uint64]struct{}, len(rels)*2)
	ids := make([]uint64, 0, len(rels)*2)
	for _, rel := range rels {
		for _, id := range []uint64{rel.SourceAssetID, rel.TargetAssetID} {
			if id == 0 {
				continue
			}
			if _, ok := seen[id]; !ok {
				seen[id] = struct{}{}
				ids = append(ids, id)
			}
		}
	}
	sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
	assets, err := s.assets.ListByIDs(ctx, ids)
	if err != nil {
		return nil, err
	}
	byID := make(map[uint64]caasmmodel.Asset, len(assets))
	for _, asset := range assets {
		byID[asset.ID] = asset
	}

	for _, id := range ids {
		asset, ok := byID[id]
		if !ok {
			// 关系指向的资产已删除/不可见：保留占位节点，保证边两端始终存在。
			graph.Nodes = append(graph.Nodes, GraphNode{
				ID:    strconv.FormatUint(id, 10),
				Label: fmt.Sprintf("asset-%d", id),
				Type:  "unknown",
			})
			continue
		}
		label := strings.TrimSpace(asset.Name)
		if label == "" {
			label = strings.TrimSpace(asset.NativeID)
		}
		if label == "" {
			label = fmt.Sprintf("asset-%d", id)
		}
		nodeType := string(asset.AssetType)
		if nodeType == "" {
			nodeType = "unknown"
		}
		graph.Nodes = append(graph.Nodes, GraphNode{
			ID:    strconv.FormatUint(id, 10),
			Label: label,
			Type:  nodeType,
		})
	}

	for _, rel := range rels {
		graph.Edges = append(graph.Edges, GraphEdge{
			Source: strconv.FormatUint(rel.SourceAssetID, 10),
			Target: strconv.FormatUint(rel.TargetAssetID, 10),
			Label:  rel.RelationshipType,
		})
	}
	return graph, nil
}

func (s *AssetService) GetDistinctServices(ctx context.Context) ([]string, error) {
	if s.assets == nil {
		return nil, errors.New("asset repository is not configured")
	}
	return s.assets.GetDistinctServices(ctx)
}

func (s *AssetService) GetDistinctEnvironments(ctx context.Context) ([]string, error) {
	if s.assets == nil {
		return nil, errors.New("asset repository is not configured")
	}
	return s.assets.GetDistinctEnvironments(ctx)
}

func (s *AssetService) UpdateOwner(ctx context.Context, actorUserID uint64, id uint64, ownerEmployeeID *uint64, departmentID *string) error {
	if s.assets == nil {
		return errors.New("asset repository is not configured")
	}
	actor, err := s.resolveActor(ctx, actorUserID)
	if err != nil {
		return err
	}
	if !s.canSetOwner(actor, ownerEmployeeID) {
		return errors.New("not authorized to update asset owner")
	}
	asset, err := s.assets.FindByID(ctx, id)
	if err != nil {
		return err
	}
	if asset == nil {
		return errors.New("asset not found")
	}
	return s.assets.UpdateAssetOwner(ctx, asset.ID, ownerEmployeeID, departmentID)
}

type PortService struct {
	ports repository.IPPortIndexRepository
}

type PortListResult struct {
	Items []caasmmodel.IPPortIndex
	Total int64
}

type PortStatsResult struct {
	Total        int64                      `json:"total"`
	Public       int64                      `json:"public"`
	Internal     int64                      `json:"internal"`
	ByStatus     map[string]int64           `json:"by_status"`
	TopOpenPorts []repository.PortFrequency `json:"top_open_ports"`
	LastUpdated  time.Time                  `json:"last_updated"`
}

func NewPortService(repo repository.IPPortIndexRepository) *PortService {
	return &PortService{ports: repo}
}

func (s *PortService) List(ctx context.Context, filter repository.PortQueryFilter, page, pageSize int) (*PortListResult, error) {
	if s.ports == nil {
		return &PortListResult{}, nil
	}
	items, total, err := s.ports.Query(ctx, filter, page, pageSize)
	if err != nil {
		return nil, err
	}
	return &PortListResult{Items: items, Total: total}, nil
}

func (s *PortService) Detail(ctx context.Context, id uint64) (*repository.PortDetail, error) {
	if s.ports == nil {
		return nil, errors.New("port repository is not configured")
	}
	return s.ports.Detail(ctx, id)
}

func (s *PortService) Stats(ctx context.Context) (*PortStatsResult, error) {
	if s.ports == nil {
		return &PortStatsResult{}, nil
	}
	stats, err := s.ports.Stats(ctx, 10)
	if err != nil {
		return nil, err
	}
	return &PortStatsResult{
		Total:        stats.Total,
		Public:       stats.Public,
		Internal:     stats.Internal,
		ByStatus:     stats.ByStatus,
		TopOpenPorts: stats.TopOpenPorts,
		LastUpdated:  stats.LastUpdated,
	}, nil
}

func (s *PortService) GetHighRiskPorts(ctx context.Context) ([]caasmmodel.IPPortIndex, error) {
	if s.ports == nil {
		return []caasmmodel.IPPortIndex{}, nil
	}
	// Common high-risk ports: FTP, SSH, Telnet, SMTP, RPC, NetBIOS, SMB, MSSQL, MySQL, RDP, PostgreSQL, Redis, Elasticsearch, MongoDB
	riskyPorts := []uint32{21, 22, 23, 25, 135, 139, 445, 1433, 3306, 3389, 5432, 6379, 9200, 27017}
	return s.ports.HighRiskPorts(ctx, riskyPorts)
}

func (s *PortService) Export(ctx context.Context) ([]string, error) {
	if s.ports == nil {
		return []string{}, nil
	}
	entries, err := s.ports.Export(ctx)
	if err != nil {
		return nil, err
	}
	lines := make([]string, 0, len(entries))
	for _, entry := range entries {
		address := strings.TrimSpace(entry.IPAddress)
		if address == "" {
			continue
		}
		lines = append(lines, formatPortExportLine(address, entry.Port))
	}
	return lines, nil
}

func formatPortExportLine(address string, port uint32) string {
	if strings.Contains(address, ":") && net.ParseIP(address) != nil {
		return fmt.Sprintf("[%s]:%d", address, port)
	}
	return fmt.Sprintf("%s:%d", address, port)
}

type DomainService struct {
	domains repository.DomainRepository
	assets  repository.AssetRepository
	users   corerepo.UserRepository
}

type DomainListResult struct {
	Items []caasmmodel.DomainAsset
	Total int64
}

type DomainStatsResult struct {
	Total        int64 `json:"total"`
	ExpiringSoon int64 `json:"expiring_soon"`
	Expired      int64 `json:"expired"`
}

func NewDomainService(repo repository.DomainRepository, assetRepo repository.AssetRepository, userRepo corerepo.UserRepository) *DomainService {
	return &DomainService{domains: repo, assets: assetRepo, users: userRepo}
}

func (s *DomainService) List(ctx context.Context, filter repository.DomainFilter, page, pageSize int) (*DomainListResult, error) {
	if s.domains == nil {
		return &DomainListResult{}, nil
	}
	items, total, err := s.domains.List(ctx, filter, page, pageSize)
	if err != nil {
		return nil, err
	}
	return &DomainListResult{Items: items, Total: total}, nil
}

func (s *DomainService) Stats(ctx context.Context) (*DomainStatsResult, error) {
	if s.domains == nil {
		return &DomainStatsResult{}, nil
	}
	stats, err := s.domains.Stats(ctx, 30*24*time.Hour)
	if err != nil {
		return nil, err
	}
	return &DomainStatsResult{
		Total:        stats.Total,
		ExpiringSoon: stats.ExpiringSoon,
		Expired:      stats.Expired,
	}, nil
}

func (s *DomainService) Claim(ctx context.Context, actorUserID uint64, id uint64, ownerEmployeeID *uint64, departmentID *string) error {
	if s.domains == nil || s.assets == nil {
		return errors.New("domain or asset repository is not configured")
	}
	if ownerEmployeeID == nil {
		return errors.New("domain owner is required")
	}
	actor, err := s.resolveActor(ctx, actorUserID)
	if err != nil {
		return err
	}
	if !s.canSetOwner(actor, ownerEmployeeID) {
		return errors.New("not authorized to claim domain owner")
	}

	domain, err := s.domains.FindByID(ctx, id)
	if err != nil {
		return err
	}

	if domain == nil {
		return errors.New("domain not found")
	}
	return s.assets.UpdateDomainOwner(ctx, domain.ID, ownerEmployeeID, departmentID)
}

type DNSRecordService struct {
	records  repository.DNSRecordRepository
	comments repository.DNSRecordCommentRepository
	users    corerepo.UserRepository
}

type DNSRecordListResult struct {
	Items []caasmmodel.DNSRecord
	Total int64
}

func NewDNSRecordService(repo repository.DNSRecordRepository, comments repository.DNSRecordCommentRepository, userRepo corerepo.UserRepository) *DNSRecordService {
	return &DNSRecordService{records: repo, comments: comments, users: userRepo}
}

func (s *DNSRecordService) List(ctx context.Context, filter repository.DNSRecordFilter, page, pageSize int) (*DNSRecordListResult, error) {
	if s.records == nil {
		return &DNSRecordListResult{}, nil
	}
	items, total, err := s.records.List(ctx, filter, page, pageSize)
	if err != nil {
		return nil, err
	}
	return &DNSRecordListResult{Items: items, Total: total}, nil
}

func (s *DNSRecordService) UpdateOwner(ctx context.Context, actorUserID uint64, id uint64, ownerEmail, deptID string, employeeID *uint64, ignoreChanges *bool, suppressPattern *string) error {
	if s.records == nil {
		return errors.New("dns record repository not configured")
	}
	actor, err := s.resolveActor(ctx, actorUserID)
	if err != nil {
		return err
	}
	if !s.canSetOwner(actor, employeeID) {
		return errors.New("not authorized to update dns record owner")
	}
	return s.records.UpdateOwner(ctx, id, ownerEmail, deptID, employeeID, ignoreChanges, suppressPattern)
}

type caasmActor struct {
	EmployeeID   *uint64
	IsPrivileged bool
}

func (s *AssetService) resolveActor(ctx context.Context, actorUserID uint64) (*caasmActor, error) {
	return resolveCAASMActor(ctx, s.users, actorUserID)
}

func (s *DomainService) resolveActor(ctx context.Context, actorUserID uint64) (*caasmActor, error) {
	return resolveCAASMActor(ctx, s.users, actorUserID)
}

func (s *DNSRecordService) resolveActor(ctx context.Context, actorUserID uint64) (*caasmActor, error) {
	return resolveCAASMActor(ctx, s.users, actorUserID)
}

func resolveCAASMActor(ctx context.Context, userRepo corerepo.UserRepository, actorUserID uint64) (*caasmActor, error) {
	if actorUserID == 0 {
		return nil, errors.New("actor required")
	}
	if userRepo == nil {
		return nil, errors.New("user repository is not configured")
	}
	user, err := userRepo.FindByID(ctx, actorUserID)
	if err != nil {
		return nil, err
	}
	if user == nil {
		return nil, errors.New("actor required")
	}
	return &caasmActor{
		EmployeeID:   user.EmployeeID,
		IsPrivileged: user.HasRole("admin") || user.HasRole("security") || user.HasRole("security_operator"),
	}, nil
}

func (s *AssetService) canSetOwner(actor *caasmActor, ownerEmployeeID *uint64) bool {
	return canCAASMSetOwner(actor, ownerEmployeeID)
}

func (s *DomainService) canSetOwner(actor *caasmActor, ownerEmployeeID *uint64) bool {
	return canCAASMSetOwner(actor, ownerEmployeeID)
}

func (s *DNSRecordService) canSetOwner(actor *caasmActor, ownerEmployeeID *uint64) bool {
	return canCAASMSetOwner(actor, ownerEmployeeID)
}

func canCAASMSetOwner(actor *caasmActor, ownerEmployeeID *uint64) bool {
	if actor == nil {
		return false
	}
	if actor.IsPrivileged {
		return true
	}
	return actor.EmployeeID != nil && ownerEmployeeID != nil && *actor.EmployeeID == *ownerEmployeeID
}

func (s *DNSRecordService) ListComments(ctx context.Context, recordID uint64) ([]caasmmodel.DNSRecordComment, error) {
	if s.comments == nil {
		return []caasmmodel.DNSRecordComment{}, nil
	}
	return s.comments.List(ctx, recordID)
}

func (s *DNSRecordService) AddComment(ctx context.Context, recordID uint64, authorName, content string) (*caasmmodel.DNSRecordComment, error) {
	if s.comments == nil {
		return nil, errors.New("dns record comment repository not configured")
	}
	comment := &caasmmodel.DNSRecordComment{
		DNSRecordID: recordID,
		Content:     content,
		AuthorName:  authorName,
	}
	if err := s.comments.Create(ctx, comment); err != nil {
		return nil, err
	}
	return comment, nil
}

type CloudAccountService struct {
	accounts   repository.CloudAccountRepository
	jobs       repository.SyncJobRepository
	assets     repository.AssetRepository
	ports      repository.IPPortIndexRepository
	history    repository.AssetHistoryRepository
	tags       repository.AssetTagRepository
	events     repository.SyncJobEventRepository
	queue      caasmqueue.JobQueueInterface
	config     *ConfigService
	dnsRecords repository.DNSRecordRepository
	logger     *logger.Logger

	// in-process guard to avoid concurrent inline syncs per account
	lockMu            sync.Mutex
	activeAccountRun  map[uint64]bool
	activeSyncCancels map[uint64]context.CancelFunc
	backoffUntil      map[uint64]time.Time

	// throttling defaults applied to collectors
	collectorFactory *collectorFactory
	collectorBuilder collectorBuilderFunc
	awsConfigBuilder func(ctx context.Context, roleARN, externalID, region string) (aws.Config, error)
	identityVerifier func(ctx context.Context, cfg aws.Config) (accountID string, arn string, err error)
}

type collectorBuilderFunc func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error)

func (s *CloudAccountService) SetEventRepository(repo repository.SyncJobEventRepository) {
	s.events = repo
}

// FindByExternalID returns a cloud account by provider + external id.
func (s *CloudAccountService) FindByExternalID(ctx context.Context, provider caasmmodel.CloudProvider, externalID string) (*caasmmodel.CloudAccount, error) {
	if s.accounts == nil {
		return nil, errors.New("cloud account repository is not configured")
	}
	return s.accounts.FindByExternalID(ctx, provider, externalID)
}

type CloudAccountListResult struct {
	Items []caasmmodel.CloudAccount
	Total int64
}

type CloudAccountTestResult struct {
	Success   bool      `json:"success"`
	Message   string    `json:"message"`
	CheckedAt time.Time `json:"checked_at"`
}

func NewCloudAccountService(repo repository.CloudAccountRepository, jobs repository.SyncJobRepository, assets repository.AssetRepository) *CloudAccountService {
	return &CloudAccountService{
		accounts:          repo,
		jobs:              jobs,
		assets:            assets,
		activeAccountRun:  make(map[uint64]bool),
		activeSyncCancels: make(map[uint64]context.CancelFunc),
		backoffUntil:      make(map[uint64]time.Time),
		collectorFactory:  newCollectorFactory(),
	}
}

func (s *CloudAccountService) SetJobQueue(jobQueue caasmqueue.JobQueueInterface) {
	s.queue = jobQueue
}

func (s *CloudAccountService) SetLogger(log *logger.Logger) {
	s.logger = log
}

func (s *CloudAccountService) SetAWSIdentityVerifier(verifier func(ctx context.Context, cfg aws.Config) (accountID string, arn string, err error)) {
	s.identityVerifier = verifier
}

func (s *CloudAccountService) SetAWSConfigBuilder(builder func(ctx context.Context, roleARN, externalID, region string) (aws.Config, error)) {
	s.awsConfigBuilder = builder
}

func (s *CloudAccountService) logInfo(message string, args ...any) {
	if s.logger != nil {
		s.logger.Info(message, args...)
	}
}

func (s *CloudAccountService) logWarn(message string, args ...any) {
	if s.logger != nil {
		s.logger.Warn(message, args...)
	}
}

func (s *CloudAccountService) logError(message string, args ...any) {
	if s.logger != nil {
		s.logger.Error(message, args...)
	}
}

func (s *CloudAccountService) SetHistoryRepository(history repository.AssetHistoryRepository) {
	s.history = history
}

func (s *CloudAccountService) SetTagRepository(tagRepo repository.AssetTagRepository) {
	s.tags = tagRepo
}

func (s *CloudAccountService) SetPortRepository(portRepo repository.IPPortIndexRepository) {
	s.ports = portRepo
}

func (s *CloudAccountService) SetDNSRecordRepository(repo repository.DNSRecordRepository) {
	s.dnsRecords = repo
}

func (s *CloudAccountService) getGoDaddyCredentials(ctx context.Context, account *caasmmodel.CloudAccount) (string, string, error) {
	if account == nil {
		return "", "", fmt.Errorf("account is nil")
	}
	meta := account.Metadata
	// Dev/test fallback: plaintext api_key/api_secret
	apiKey, _ := meta["api_key"].(string)
	apiSecret, _ := meta["api_secret"].(string)

	// Secrets Manager path
	if arn, ok := meta["secret_arn"].(string); ok && arn != "" {
		cfg, err := config.LoadDefaultConfig(ctx)
		if err != nil {
			return "", "", fmt.Errorf("load aws config for secrets manager: %w", err)
		}
		sm := secretsmanager.NewFromConfig(cfg)
		out, err := sm.GetSecretValue(ctx, &secretsmanager.GetSecretValueInput{
			SecretId: aws.String(arn),
		})
		if err != nil {
			return "", "", fmt.Errorf("get secret %s: %w", arn, err)
		}
		if out.SecretString == nil {
			return "", "", fmt.Errorf("secret %s has no string value", arn)
		}
		var payload map[string]string
		if err := json.Unmarshal([]byte(*out.SecretString), &payload); err != nil {
			return "", "", fmt.Errorf("parse secret payload: %w", err)
		}
		if v, ok := payload["api_key"]; ok && v != "" {
			apiKey = v
		}
		if v, ok := payload["api_secret"]; ok && v != "" {
			apiSecret = v
		}
	}

	if apiKey == "" || apiSecret == "" {
		return "", "", fmt.Errorf("missing GoDaddy credentials (api_key/api_secret)")
	}
	return apiKey, apiSecret, nil
}

// CreateProviderSecret is disabled with the retired GoDaddy CAASM integration.
func (s *CloudAccountService) CreateProviderSecret(ctx context.Context, provider caasmmodel.CloudProvider, name, region, apiKey, apiSecret string) (string, error) {
	return "", errors.New("GoDaddy CAASM collection is disabled")
}

func (s *CloudAccountService) SetConfigService(svc *ConfigService) {
	s.config = svc
}

func (s *CloudAccountService) buildCollector(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
	if s.collectorBuilder != nil {
		return s.collectorBuilder(ctx, account)
	}
	return s.buildDefaultCollector(ctx, account)
}

func (s *CloudAccountService) buildDefaultCollector(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
	if account == nil {
		return nil, "", errors.New("cloud account not found")
	}

	meta := account.Metadata
	roleARN, _ := meta["role_arn"].(string)
	externalID, _ := meta["external_id"].(string)
	region, _ := meta["region"].(string)

	var allowedRegions []string
	if regionsInterface, ok := meta["allowed_regions"]; ok {
		if regionsList, ok := regionsInterface.([]interface{}); ok {
			for _, r := range regionsList {
				if rStr, ok := r.(string); ok {
					allowedRegions = append(allowedRegions, rStr)
				}
			}
		}
	}

	var (
		col collector.Collector
		err error
	)

	if account.Provider == "aws" {
		cfg, cfgErr := s.assumeRoleAWS(ctx, roleARN, externalID, region)
		if cfgErr != nil {
			return nil, region, cfgErr
		}
		actualAccountID, actualARN, identityErr := s.verifyAWSIdentity(ctx, cfg)
		if identityErr != nil {
			return nil, region, identityErr
		}
		if identityErr := validateAWSAccountIdentity(account, roleARN, actualAccountID, actualARN); identityErr != nil {
			return nil, region, identityErr
		}
		col, err = aws_collector.NewAWSCollectorWithConfig(cfg, region, uint(account.ID), allowedRegions)
		if err != nil {
			return nil, region, err
		}
		if awsCol, ok := col.(*aws_collector.AWSCollector); ok {
			scope := s.resolveCollectionScope(ctx)
			awsCol.SetDomainCollectionScope(scope.DomainScope)
			awsCol.SetExecutionGate(func(execCtx context.Context) error {
				return s.blockIfKillSwitchEnabled(execCtx, nil, account.ID, "aws_api_gate")
			})
		}
	} else if account.Provider == caasmmodel.CloudProviderGoDaddy {
		/*
			GoDaddy CAASM collection is disabled. The collector code remains in the
			repository for historical reference, but it is no longer wired into runtime
			sync because the feature is no longer used.

			apiKey, apiSecret, credErr := s.getGoDaddyCredentials(ctx, account)
			if credErr != nil {
				return nil, region, credErr
			}
			gd := godaddycollector.NewCollector(apiKey, apiSecret, account.ID)
			if env, ok := account.Metadata["environment"].(string); ok {
				gd.SetEnvironment(env)
			}
			if valErr := gd.Validate(); valErr != nil {
				return nil, region, valErr
			}
			col = gd
		*/
		return nil, region, errors.New("GoDaddy CAASM collection is disabled")
	} else {
		return nil, region, fmt.Errorf("unsupported provider: %s", account.Provider)
	}

	if awsCol, ok := col.(*aws_collector.AWSCollector); ok {
		if s.collectorFactory != nil {
			s.collectorFactory.ApplyThrottling(awsCol)
		}

		base := ResolveGlobalThrottling(nil).Effective
		if s.collectorFactory != nil {
			base = s.collectorFactory.CurrentThrottling()
		}
		effective := ResolveAccountThrottling(base, account).Effective

		awsCol.SetThrottleEnabled(effective.Enabled)
		awsCol.ConfigureRateLimiter(effective.TokenCapacity, effective.TokenRefillPerSec)
		awsCol.ConfigureAdaptive(
			effective.AdaptiveThreshold,
			time.Duration(effective.AdaptiveWindowSec)*time.Second,
			time.Duration(effective.CoolOffSec)*time.Second,
			effective.MaxConcurrency,
			effective.MinConcurrency,
		)
	}

	return col, region, nil
}

func (s *CloudAccountService) verifyAWSIdentity(ctx context.Context, cfg aws.Config) (string, string, error) {
	if s.identityVerifier != nil {
		return s.identityVerifier(ctx, cfg)
	}
	stsClient := sts.NewFromConfig(cfg)
	out, err := stsClient.GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{})
	if err != nil {
		return "", "", fmt.Errorf("STS GetCallerIdentity failed: %w", err)
	}
	return aws.ToString(out.Account), aws.ToString(out.Arn), nil
}

func (s *CloudAccountService) isBackoffActive(accountID uint64) bool {
	s.lockMu.Lock()
	defer s.lockMu.Unlock()
	until, ok := s.backoffUntil[accountID]
	return ok && time.Now().Before(until)
}

func (s *CloudAccountService) setBackoff(accountID uint64, base time.Duration) {
	jitter := time.Duration(rand.Intn(15)) * time.Second
	delay := base + jitter
	if delay > 10*time.Minute {
		delay = 10 * time.Minute
	}
	s.lockMu.Lock()
	s.backoffUntil[accountID] = time.Now().Add(delay)
	s.lockMu.Unlock()
}

func (s *CloudAccountService) clearBackoff(accountID uint64) {
	s.lockMu.Lock()
	delete(s.backoffUntil, accountID)
	s.lockMu.Unlock()
}

// GetBackoffUntil returns the backoff expiration time for an account, or zero time if no backoff is active.
func (s *CloudAccountService) GetBackoffUntil(accountID uint64) time.Time {
	s.lockMu.Lock()
	defer s.lockMu.Unlock()
	if until, ok := s.backoffUntil[accountID]; ok {
		return until
	}
	return time.Time{}
}

// ClearBackoff manually clears the backoff for an account (admin override).
func (s *CloudAccountService) ClearBackoff(accountID uint64) {
	s.clearBackoff(accountID)
}

// tryLockAccount returns true if it acquired an in-process lock for the given account ID.
func (s *CloudAccountService) tryLockAccount(accountID uint64) bool {
	s.lockMu.Lock()
	defer s.lockMu.Unlock()
	if s.activeAccountRun[accountID] {
		return false
	}
	s.activeAccountRun[accountID] = true
	return true
}

func (s *CloudAccountService) unlockAccount(accountID uint64) {
	s.lockMu.Lock()
	delete(s.activeAccountRun, accountID)
	s.lockMu.Unlock()
}

func (s *CloudAccountService) registerSyncCancel(syncJobID uint64, cancel context.CancelFunc) {
	if syncJobID == 0 || cancel == nil {
		return
	}
	s.lockMu.Lock()
	s.activeSyncCancels[syncJobID] = cancel
	s.lockMu.Unlock()
}

func (s *CloudAccountService) clearSyncCancel(syncJobID uint64) {
	if syncJobID == 0 {
		return
	}
	s.lockMu.Lock()
	delete(s.activeSyncCancels, syncJobID)
	s.lockMu.Unlock()
}

func (s *CloudAccountService) resolveExecutionTimeoutMinutes() int {
	timeoutMinutes := caasmmodel.DefaultSyncJobExecutionTimeoutMinutes
	if s == nil || s.config == nil {
		return timeoutMinutes
	}
	settings, err := s.config.ResolveSyncSettings(context.Background())
	if err != nil {
		return timeoutMinutes
	}
	if settings.ExecutionTimeoutMinutes > 0 {
		timeoutMinutes = settings.ExecutionTimeoutMinutes
	}
	return timeoutMinutes
}

func deriveCancellationSource(ctx context.Context, contextSource string) string {
	if ctx == nil || !errors.Is(ctx.Err(), context.Canceled) {
		return ""
	}
	switch strings.TrimSpace(strings.ToLower(contextSource)) {
	case "queue_worker", "serverless_simulated":
		return "parent_context_canceled"
	default:
		return "unknown_external"
	}
}

func errString(err error, fallback string) string {
	if err != nil {
		return err.Error()
	}
	return fallback
}

func (s *CloudAccountService) runSyncWithExecutionContext(
	parentCtx context.Context,
	job *caasmmodel.SyncJob,
	account *caasmmodel.CloudAccount,
	contextSource string,
) {
	if parentCtx == nil {
		parentCtx = context.Background()
	}
	if contextSource == "" {
		contextSource = "unknown_execution_path"
	}
	timeoutMinutes := s.resolveExecutionTimeoutMinutes()
	execCtx, cancel := context.WithTimeout(parentCtx, time.Duration(timeoutMinutes)*time.Minute)
	defer cancel()

	if job != nil {
		s.registerSyncCancel(job.ID, cancel)
		defer s.clearSyncCancel(job.ID)
		if job.Metadata == nil {
			job.Metadata = coremodel.JSONB{}
		}
		job.Metadata["context_source"] = contextSource
		job.Metadata["execution_timeout_minutes"] = timeoutMinutes
		if s.jobs != nil {
			_ = s.jobs.UpdateFields(context.Background(), job.ID, map[string]interface{}{
				"metadata": job.Metadata,
			})
		}
	}

	s.runSync(execCtx, job, account)
}

func (s *CloudAccountService) CancelSyncJob(syncJobID uint64) bool {
	if syncJobID == 0 {
		return false
	}
	s.lockMu.Lock()
	cancel, ok := s.activeSyncCancels[syncJobID]
	s.lockMu.Unlock()
	if !ok || cancel == nil {
		return false
	}
	cancel()
	return true
}

func (s *CloudAccountService) GetTags(ctx context.Context, assetID uint64) (map[string]string, error) {
	if s.tags == nil {
		return map[string]string{}, nil
	}
	return s.tags.GetTags(ctx, assetID)
}

func (s *CloudAccountService) ReplaceTags(ctx context.Context, assetID uint64, tags map[string]string) error {
	if s.tags == nil {
		return errors.New("tag repository is not configured")
	}
	before, _ := s.tags.GetTags(ctx, assetID)
	if err := s.tags.ReplaceTags(ctx, assetID, tags); err != nil {
		return err
	}
	change := coremodel.JSONB{
		"before": before,
		"after":  tags,
	}
	s.recordHistory(ctx, assetID, "tags.update", change)
	return nil
}
func (s *CloudAccountService) List(ctx context.Context, filter repository.CloudAccountFilter, page, pageSize int) (*CloudAccountListResult, error) {
	if s.accounts == nil {
		return &CloudAccountListResult{}, nil
	}
	items, total, err := s.accounts.List(ctx, filter, page, pageSize)
	if err != nil {
		return nil, err
	}
	return &CloudAccountListResult{Items: items, Total: total}, nil
}

func (s *CloudAccountService) Get(ctx context.Context, id uint64) (*caasmmodel.CloudAccount, error) {
	if s.accounts == nil {
		return nil, errors.New("cloud account repository is not configured")
	}
	return s.accounts.FindByID(ctx, id)
}

func (s *CloudAccountService) Create(ctx context.Context, account *caasmmodel.CloudAccount) error {
	if s.accounts == nil {
		return errors.New("cloud account repository is not configured")
	}
	if err := s.validateCloudAccountInput(ctx, account); err != nil {
		return err
	}
	return s.accounts.Create(ctx, account)
}

func (s *CloudAccountService) Update(ctx context.Context, account *caasmmodel.CloudAccount) error {
	if s.accounts == nil {
		return errors.New("cloud account repository is not configured")
	}
	if err := s.validateCloudAccountInput(ctx, account); err != nil {
		return err
	}
	return s.accounts.Update(ctx, account)
}

func (s *CloudAccountService) Delete(ctx context.Context, id uint64) error {
	if s.accounts == nil {
		return errors.New("cloud account repository is not configured")
	}
	return s.accounts.Delete(ctx, id)
}

func (s *CloudAccountService) Test(ctx context.Context, id uint64) (*CloudAccountTestResult, error) {
	account, err := s.Get(ctx, id)
	if err != nil {
		return nil, err
	}
	return s.validateAccount(ctx, account)
}

// validateAccount performs provider-specific credential checks.
func (s *CloudAccountService) validateAccount(ctx context.Context, account *caasmmodel.CloudAccount) (*CloudAccountTestResult, error) {
	if account == nil {
		return nil, errors.New("account is nil")
	}

	switch account.Provider {
	case caasmmodel.CloudProviderAWS:
		return s.validateAWS(ctx, account)
	default:
		return &CloudAccountTestResult{
			Success:   false,
			Message:   fmt.Sprintf("provider %s not supported for validation", account.Provider),
			CheckedAt: time.Now(),
		}, nil
	}
}

func (s *CloudAccountService) validateAWS(ctx context.Context, account *caasmmodel.CloudAccount) (*CloudAccountTestResult, error) {
	meta := account.Metadata
	roleARN, _ := meta["role_arn"].(string)
	externalID, _ := meta["external_id"].(string)
	region, _ := meta["region"].(string)

	cfg, err := s.assumeRoleAWS(ctx, roleARN, externalID, region)
	if err != nil {
		return &CloudAccountTestResult{
			Success:   false,
			Message:   fmt.Sprintf("failed to assume role: %v", err),
			CheckedAt: time.Now(),
		}, nil
	}

	actualAccountID, actualARN, err := s.verifyAWSIdentity(ctx, cfg)
	if err != nil {
		return &CloudAccountTestResult{
			Success:   false,
			Message:   err.Error(),
			CheckedAt: time.Now(),
		}, nil
	}
	if err := validateAWSAccountIdentity(account, roleARN, actualAccountID, actualARN); err != nil {
		return &CloudAccountTestResult{
			Success:   false,
			Message:   err.Error(),
			CheckedAt: time.Now(),
		}, nil
	}

	// Optionally record last_sync_at/error_message for visibility
	now := time.Now()
	account.LastSyncAt = &now
	account.ErrorMessage = ""
	_ = s.accounts.Update(ctx, account)

	return &CloudAccountTestResult{
		Success:   true,
		Message:   fmt.Sprintf("validated AWS account: %s", actualARN),
		CheckedAt: time.Now(),
	}, nil
}

var (
	awsAccountIDPattern = regexp.MustCompile(`^\d{12}$`)
	awsRoleARNPattern   = regexp.MustCompile(`^arn:aws[a-zA-Z-]*:iam::\d{12}:role\/.+$`)
	awsRegionPattern    = regexp.MustCompile(`^[a-z]{2}(-gov)?-[a-z0-9-]+-\d+$`)
)

func (s *CloudAccountService) validateCloudAccountInput(ctx context.Context, account *caasmmodel.CloudAccount) error {
	if account == nil {
		return errors.New("account is nil")
	}

	account.Name = strings.TrimSpace(account.Name)
	account.ExternalID = strings.TrimSpace(account.ExternalID)
	if account.Metadata == nil {
		account.Metadata = coremodel.JSONB{}
	}

	if account.Name == "" {
		return errors.New("name is required")
	}
	if account.Provider == "" {
		return errors.New("provider is required")
	}
	if account.Provider == caasmmodel.CloudProviderGoDaddy {
		return errors.New("GoDaddy CAASM collection is disabled")
	}
	if account.ExternalID == "" {
		return errors.New("external_id is required")
	}
	if account.Status == "" {
		account.Status = caasmmodel.CloudAccountStatusActive
	}

	switch account.Provider {
	case caasmmodel.CloudProviderAWS:
		globalThrottling, err := s.currentGlobalThrottling(ctx)
		if err != nil {
			return fmt.Errorf("resolve global throttling defaults: %w", err)
		}
		return validateAWSAccountInput(account, globalThrottling)
	default:
		return nil
	}
}

func validateAWSAccountInput(account *caasmmodel.CloudAccount, globalThrottling caasmmodel.ThrottlingConfig) error {
	if !awsAccountIDPattern.MatchString(account.ExternalID) {
		return errors.New("external_id must be a 12-digit AWS account ID")
	}

	roleARN, _ := account.Metadata["role_arn"].(string)
	roleARN = strings.TrimSpace(roleARN)
	if roleARN == "" {
		return errors.New("metadata.role_arn is required for AWS cloud accounts")
	}
	if !awsRoleARNPattern.MatchString(roleARN) {
		return errors.New("metadata.role_arn must be a valid AWS IAM role ARN")
	}
	roleAccountID := awsAccountIDFromRoleARN(roleARN)
	if roleAccountID != "" && roleAccountID != account.ExternalID {
		return errors.New("metadata.role_arn account ID must match external_id")
	}
	account.Metadata["role_arn"] = roleARN

	if externalID, ok := account.Metadata["external_id"].(string); ok {
		externalID = strings.TrimSpace(externalID)
		if externalID == "" {
			delete(account.Metadata, "external_id")
		} else {
			account.Metadata["external_id"] = externalID
		}
	}

	region, _ := account.Metadata["region"].(string)
	region = strings.TrimSpace(region)
	if region == "" {
		account.Metadata["region"] = "us-east-1"
	} else {
		account.Metadata["region"] = region
	}

	allowedRegions, err := normalizeAllowedRegions(account.Metadata["allowed_regions"])
	if err != nil {
		return err
	}
	if len(allowedRegions) == 0 {
		delete(account.Metadata, "allowed_regions")
	} else {
		account.Metadata["allowed_regions"] = allowedRegions
	}

	if v, ok := account.Metadata["throttle_enabled"]; ok {
		enabled, ok := v.(bool)
		if !ok {
			return errors.New("metadata.throttle_enabled must be a boolean")
		}
		account.Metadata["throttle_enabled"] = enabled
	}

	if value, ok, err := normalizeOptionalIntMetadata(account.Metadata["throttle_tokens"], "metadata.throttle_tokens", caasmmodel.HardCapThrottleTokenCapacity); err != nil {
		return err
	} else if ok {
		if globalThrottling.TokenCapacity > 0 && value > globalThrottling.TokenCapacity {
			return fmt.Errorf("metadata.throttle_tokens cannot exceed global throttling.token_capacity (%d)", globalThrottling.TokenCapacity)
		}
		account.Metadata["throttle_tokens"] = value
	} else {
		delete(account.Metadata, "throttle_tokens")
	}

	if value, ok, err := normalizeOptionalFloatMetadata(account.Metadata["throttle_refill_per_sec"], "metadata.throttle_refill_per_sec", float64(caasmmodel.HardCapThrottleTokenRefillPerSec)); err != nil {
		return err
	} else if ok {
		if globalThrottling.TokenRefillPerSec > 0 && value > globalThrottling.TokenRefillPerSec {
			return fmt.Errorf("metadata.throttle_refill_per_sec cannot exceed global throttling.token_refill_per_sec (%g)", globalThrottling.TokenRefillPerSec)
		}
		account.Metadata["throttle_refill_per_sec"] = value
	} else {
		delete(account.Metadata, "throttle_refill_per_sec")
	}

	maxConc, hasMax, err := normalizeOptionalIntMetadata(account.Metadata["throttle_max_concurrency"], "metadata.throttle_max_concurrency", caasmmodel.HardCapThrottleMaxConcurrency)
	if err != nil {
		return err
	}
	minConc, hasMin, err := normalizeOptionalIntMetadata(account.Metadata["throttle_min_concurrency"], "metadata.throttle_min_concurrency", caasmmodel.HardCapThrottleMinConcurrency)
	if err != nil {
		return err
	}
	if hasMax {
		if globalThrottling.MaxConcurrency > 0 && maxConc > globalThrottling.MaxConcurrency {
			return fmt.Errorf("metadata.throttle_max_concurrency cannot exceed global throttling.max_concurrency (%d)", globalThrottling.MaxConcurrency)
		}
		account.Metadata["throttle_max_concurrency"] = maxConc
	} else {
		delete(account.Metadata, "throttle_max_concurrency")
	}
	if hasMin {
		if globalThrottling.MinConcurrency > 0 && minConc > globalThrottling.MinConcurrency {
			return fmt.Errorf("metadata.throttle_min_concurrency cannot exceed global throttling.min_concurrency (%d)", globalThrottling.MinConcurrency)
		}
		account.Metadata["throttle_min_concurrency"] = minConc
	} else {
		delete(account.Metadata, "throttle_min_concurrency")
	}
	if hasMax && hasMin && minConc > maxConc {
		return errors.New("metadata.throttle_min_concurrency cannot exceed metadata.throttle_max_concurrency")
	}

	return nil
}

func awsAccountIDFromRoleARN(roleARN string) string {
	parts := strings.Split(roleARN, ":")
	if len(parts) < 5 {
		return ""
	}
	return parts[4]
}

type awsIdentityValidationError struct {
	ExpectedAccountID string
	RoleAccountID     string
	ActualAccountID   string
	ActualARN         string
}

func (e *awsIdentityValidationError) Error() string {
	return fmt.Sprintf("AWS identity account mismatch: expected %s from role account %s, got %s (%s)", e.ExpectedAccountID, e.RoleAccountID, e.ActualAccountID, e.ActualARN)
}

func validateAWSAccountIdentity(account *caasmmodel.CloudAccount, roleARN string, actualAccountID string, actualARN string) error {
	if account == nil {
		return errors.New("account is nil")
	}
	expectedAccountID := strings.TrimSpace(account.ExternalID)
	roleAccountID := awsAccountIDFromRoleARN(strings.TrimSpace(roleARN))
	actualAccountID = strings.TrimSpace(actualAccountID)
	if actualAccountID == "" {
		return &awsIdentityValidationError{
			ExpectedAccountID: expectedAccountID,
			RoleAccountID:     roleAccountID,
			ActualAccountID:   actualAccountID,
			ActualARN:         actualARN,
		}
	}
	if expectedAccountID != "" && actualAccountID != expectedAccountID {
		return &awsIdentityValidationError{
			ExpectedAccountID: expectedAccountID,
			RoleAccountID:     roleAccountID,
			ActualAccountID:   actualAccountID,
			ActualARN:         actualARN,
		}
	}
	if expectedAccountID != "" && roleAccountID != "" && roleAccountID != expectedAccountID {
		return &awsIdentityValidationError{
			ExpectedAccountID: expectedAccountID,
			RoleAccountID:     roleAccountID,
			ActualAccountID:   actualAccountID,
			ActualARN:         actualARN,
		}
	}
	return nil
}

func normalizeAllowedRegions(value interface{}) ([]string, error) {
	if value == nil {
		return nil, nil
	}

	var raw []string
	switch v := value.(type) {
	case []string:
		raw = append(raw, v...)
	case []interface{}:
		for _, item := range v {
			s, ok := item.(string)
			if !ok {
				return nil, errors.New("metadata.allowed_regions must be an array of strings")
			}
			raw = append(raw, s)
		}
	case string:
		for _, item := range strings.Split(v, ",") {
			raw = append(raw, item)
		}
	default:
		return nil, errors.New("metadata.allowed_regions must be an array of strings")
	}

	normalized := make([]string, 0, len(raw))
	seen := make(map[string]struct{}, len(raw))
	for _, region := range raw {
		trimmed := strings.ToLower(strings.TrimSpace(region))
		if trimmed == "" {
			continue
		}
		if !awsRegionPattern.MatchString(trimmed) {
			return nil, fmt.Errorf("metadata.allowed_regions contains invalid AWS region: %s", trimmed)
		}
		if _, exists := seen[trimmed]; exists {
			continue
		}
		seen[trimmed] = struct{}{}
		normalized = append(normalized, trimmed)
	}
	return normalized, nil
}

func normalizeOptionalIntMetadata(value interface{}, field string, max int) (int, bool, error) {
	if value == nil {
		return 0, false, nil
	}

	var normalized int
	switch v := value.(type) {
	case int:
		normalized = v
	case int32:
		normalized = int(v)
	case int64:
		normalized = int(v)
	case float64:
		if v != float64(int(v)) {
			return 0, false, fmt.Errorf("%s must be a whole number", field)
		}
		normalized = int(v)
	default:
		return 0, false, fmt.Errorf("%s must be a number", field)
	}

	if normalized <= 0 {
		return 0, false, fmt.Errorf("%s must be >0", field)
	}
	if normalized > max {
		return 0, false, fmt.Errorf("%s must be <=%d", field, max)
	}
	return normalized, true, nil
}

func normalizeOptionalFloatMetadata(value interface{}, field string, max float64) (float64, bool, error) {
	if value == nil {
		return 0, false, nil
	}

	var normalized float64
	switch v := value.(type) {
	case float64:
		normalized = v
	case float32:
		normalized = float64(v)
	case int:
		normalized = float64(v)
	case int32:
		normalized = float64(v)
	case int64:
		normalized = float64(v)
	default:
		return 0, false, fmt.Errorf("%s must be a number", field)
	}

	if normalized <= 0 {
		return 0, false, fmt.Errorf("%s must be >0", field)
	}
	if normalized > max {
		return 0, false, fmt.Errorf("%s must be <=%g", field, max)
	}
	return normalized, true, nil
}

// assumeRoleAWS builds an AWS config by assuming the target role using the default credential chain.
func (s *CloudAccountService) assumeRoleAWS(ctx context.Context, roleARN, externalID, region string) (aws.Config, error) {
	if s.awsConfigBuilder != nil {
		return s.awsConfigBuilder(ctx, roleARN, externalID, region)
	}
	if roleARN == "" {
		return aws.Config{}, errors.New("missing role_arn for AWS account")
	}
	if region == "" {
		region = "us-east-1"
	}
	cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(region))
	if err != nil {
		return aws.Config{}, fmt.Errorf("failed to load default AWS config: %w", err)
	}

	// Preflight source credentials so we surface expired/invalid SSO early
	sourceSTS := sts.NewFromConfig(cfg)
	if _, err := sourceSTS.GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{}); err != nil {
		if strings.Contains(err.Error(), "SSO") || strings.Contains(err.Error(), "token") {
			return aws.Config{}, fmt.Errorf("source AWS credentials are invalid or expired (SSO/session). Refresh credentials (e.g., AWS_PROFILE=<profile> aws sso login) or provide non-SSO credentials for the collector: %w", err)
		}
		return aws.Config{}, fmt.Errorf("failed to validate source AWS credentials: %w", err)
	}

	stsClient := sts.NewFromConfig(cfg)
	provider := stscreds.NewAssumeRoleProvider(stsClient, roleARN, func(o *stscreds.AssumeRoleOptions) {
		if externalID != "" {
			o.ExternalID = &externalID
		}
	})
	cfg.Credentials = aws.NewCredentialsCache(provider)
	return cfg, nil
}

func (s *CloudAccountService) TriggerSync(ctx context.Context, id uint64, actorID *uint64) (*caasmmodel.SyncJob, error) {
	return s.TriggerSyncWithPriority(ctx, id, actorID, caasmqueue.PriorityNormal)
}

// TriggerSyncWithPriority enqueues or runs a sync with explicit priority.
func (s *CloudAccountService) TriggerSyncWithPriority(ctx context.Context, id uint64, actorID *uint64, priority caasmqueue.JobPriority) (*caasmmodel.SyncJob, error) {
	if s.accounts == nil || s.jobs == nil {
		return nil, errors.New("sync pipeline is not configured")
	}
	if err := s.blockIfKillSwitchEnabled(ctx, nil, id, "admission"); err != nil {
		return nil, err
	}

	// Backoff guard
	if s.isBackoffActive(id) {
		retryAt := s.GetBackoffUntil(id)
		return nil, &BackoffError{AccountID: id, RetryAt: retryAt}
	}

	// Cross-process guard: return existing active job if one is queued/running
	if s.jobs != nil {
		activeJob, err := s.jobs.FindActiveJobForAccount(ctx, id)
		if err != nil {
			return nil, err
		}
		if activeJob != nil {
			return activeJob, nil
		}
	}

	account, job, err := s.createSyncJobRecord(ctx, id, actorID)
	if err != nil {
		return nil, err
	}

	s.recordJobEvent(ctx, job.ID, "info", "Job created", map[string]interface{}{
		"cloud_account_id": id,
		"priority":         priority,
	})
	if err := s.blockIfKillSwitchEnabled(ctx, job, id, "pre_dispatch"); err != nil {
		return job, err
	}

	// Check sync mode from config
	mode := "queue"
	var lambdaName string
	if s.config != nil {
		if settings, err := s.config.ResolveSyncSettings(ctx); err == nil {
			if settings.Mode != "" {
				mode = settings.Mode
			}
			lambdaName = settings.LambdaFunctionName
		}
		if resolution, err := s.config.ResolveEffectiveThrottling(ctx); err == nil && s.collectorFactory != nil {
			s.collectorFactory.ConfigureThrottling(resolution.Effective)
		}
	}

	payload := &caasmqueue.SyncJobPayload{
		SyncJobID:      &job.ID,
		CloudAccountID: id,
		JobType:        caasmqueue.JobTypeCloudAccountSync,
		Priority:       priority,
		TriggeredBy:    actorID,
	}
	/*
		GoDaddy sync dispatch is disabled with the retired GoDaddy CAASM integration.

		if account != nil && account.Provider == caasmmodel.CloudProviderGoDaddy {
			payload.JobType = caasmqueue.JobTypeGoDaddySync
		}
	*/

	roleARN, _ := account.Metadata["role_arn"].(string)
	externalID, _ := account.Metadata["external_id"].(string)
	_ = roleARN
	_ = externalID

	/*
		GoDaddy queue-only dispatch is disabled with the retired GoDaddy CAASM integration.

		if account != nil && account.Provider == caasmmodel.CloudProviderGoDaddy {
			mode = "queue"
			lambdaName = ""
			s.recordJobEvent(ctx, job.ID, "info", "GoDaddy jobs enforced to queue mode", map[string]interface{}{
				"job_type": payload.JobType,
			})
			if s.queue == nil {
				job.Status = caasmmodel.SyncJobFailed
				job.ErrorMessage = "queue is required for GoDaddy sync but queue is not configured"
				_ = s.jobs.UpdateFields(ctx, job.ID, map[string]interface{}{
					"status":        job.Status,
					"error_message": job.ErrorMessage,
				})
				s.recordJobEvent(ctx, job.ID, "error", job.ErrorMessage, nil)
				return job, errors.New(job.ErrorMessage)
			}
		}
	*/

	if mode == "serverless" {
		// NOTE: In the current architecture, "serverless" here is a local/service-side
		// execution mode label, not a real AWS Lambda invoke path. The API service still
		// dispatches through invokeLambdaSimulated(...) below, which reuses the in-process
		// ExecuteSyncPayload path asynchronously. A separate deployable Lambda handler
		// exists under cmd/lambda/collector, but this service path does not call it.
		// Mark job as enqueued for serverless execution before invoking the worker.
		// This prevents a race condition where the worker (or simulated goroutine) reads the job
		// before the metadata is updated.
		job.Status = caasmmodel.SyncJobQueued
		// Create a new metadata map to ensure GORM detects the change
		job.Metadata = coremodel.JSONB{
			"account_name":   job.Metadata["account_name"],
			"provider":       job.Metadata["provider"],
			"execution_mode": "serverless",
		}
		// Use Updates with a map to explicitly specify which fields to update
		// This ensures GORM updates the JSONB metadata field
		updateErr := s.jobs.UpdateFields(ctx, job.ID, map[string]interface{}{
			"status":   caasmmodel.SyncJobQueued,
			"metadata": job.Metadata,
		})

		if updateErr != nil {
			// If we fail to update metadata, we should probably fail the job or fallback.
			// For now, we'll log and continue, but in reality we might want to abort.
		}

		if lambdaName == "" {
			// Misconfigured serverless mode; run inline to avoid enqueue dependency.
			s.recordJobEvent(ctx, job.ID, "warning", "Serverless mode missing worker label - running inline", map[string]interface{}{
				"mode":     mode,
				"priority": priority,
			})
			s.runSyncWithExecutionContext(context.Background(), job, account, "serverless_inline_fallback")
			return job, nil
		}

		// Fire-and-forget invocation so the API returns immediately (avoid client timeouts).
		go func(jobID uint64, lambda string, p caasmqueue.SyncJobPayload) {
			defer func() {
				if r := recover(); r != nil {
					s.logError("serverless worker invocation panicked",
						"job_id", jobID,
						"worker_label", lambda,
						"mode", mode,
						"priority", priority,
						"error", fmt.Sprintf("%v", r),
					)
					s.recordJobEvent(ctx, jobID, "error", "Serverless worker invocation panic", map[string]interface{}{
						"worker_label": lambda,
						"mode":         mode,
						"priority":     priority,
						"error":        fmt.Sprintf("%v", r),
					})
				}
			}()
			invokeErr := s.invokeLambdaSimulated(context.Background(), lambda, &p)
			if invokeErr != nil {
				s.recordJobEvent(ctx, jobID, "error", "Serverless worker invocation failed", map[string]interface{}{
					"worker_label": lambda,
					"mode":         mode,
					"priority":     priority,
					"error":        invokeErr.Error(),
				})
				// Fallback inline to avoid losing the job entirely.
				s.runSyncWithExecutionContext(context.Background(), job, account, "serverless_invoke_fallback")
			}
		}(job.ID, lambdaName, *payload)

		s.recordJobEvent(ctx, job.ID, "info", "Serverless worker invocation requested", map[string]interface{}{
			"worker_label": lambdaName,
			"mode":         mode,
			"priority":     priority,
		})

		return job, nil
	}

	// Queue mode (or fallback)
	if s.queue != nil {
		if err := s.queue.Enqueue(ctx, payload); err != nil {
			// Mark the job as failed to reflect the enqueue error
			s.recordJobEvent(ctx, job.ID, "error", "Queue enqueue failed - running inline fallback", map[string]interface{}{
				"mode":     mode,
				"priority": priority,
				"error":    err.Error(),
			})
			s.runSyncWithExecutionContext(context.Background(), job, account, "queue_inline_fallback")
			return job, nil
		}
		s.recordJobEvent(ctx, job.ID, "info", "Job enqueued to Redis queue", map[string]interface{}{
			"priority": priority,
			"mode":     mode,
		})
		return job, nil
	}

	s.recordJobEvent(ctx, job.ID, "info", "Job running synchronously (no queue available)", map[string]interface{}{
		"priority": priority,
		"mode":     mode,
	})
	s.runSyncWithExecutionContext(context.Background(), job, account, "direct_inline")
	return job, nil
}

func (s *CloudAccountService) createSyncJobRecord(ctx context.Context, cloudAccountID uint64, actorID *uint64) (*caasmmodel.CloudAccount, *caasmmodel.SyncJob, error) {
	account, err := s.accounts.FindByID(ctx, cloudAccountID)
	if err != nil {
		return nil, nil, err
	}
	if account.Provider == caasmmodel.CloudProviderGoDaddy {
		return nil, nil, errors.New("GoDaddy CAASM collection is disabled")
	}
	// Normalize actor: zero is treated as anonymous to avoid FK violations.
	if actorID != nil && *actorID == 0 {
		actorID = nil
	}

	jobType := "cloud_account_sync"
	/*
		GoDaddy job creation is disabled before this point.

		if account.Provider == caasmmodel.CloudProviderGoDaddy {
			jobType = "godaddy_sync"
		}
	*/
	job := &caasmmodel.SyncJob{
		JobType:        jobType,
		Status:         caasmmodel.SyncJobQueued,
		CloudAccountID: &cloudAccountID,
		TriggeredBy:    actorID,
		Metadata: coremodel.JSONB{
			"account_name": account.Name,
			"provider":     account.Provider,
		},
	}
	if err := s.jobs.Create(ctx, job); err != nil {
		return nil, nil, err
	}
	return account, job, nil
}

func (s *CloudAccountService) runSync(ctx context.Context, job *caasmmodel.SyncJob, account *caasmmodel.CloudAccount) {
	if account == nil {
		if job != nil && s.jobs != nil {
			job.Status = caasmmodel.SyncJobFailed
			job.ErrorMessage = "cloud account not found"
			_ = s.jobs.UpdateFields(ctx, job.ID, map[string]interface{}{
				"status":        job.Status,
				"error_message": job.ErrorMessage,
			})
		}
		return
	}
	if err := s.blockIfKillSwitchEnabled(ctx, job, account.ID, "sync_start"); err != nil {
		return
	}

	locked := false
	locked = s.tryLockAccount(account.ID)
	if !locked {
		// Mark job as skipped to reflect locking. Persist the guard metadata in the
		// same write as the terminal status to avoid losing the signal on fast-fail paths.
		job.Status = caasmmodel.SyncJobFailed
		job.ErrorMessage = "another sync is already running for this account"
		if job.Metadata == nil {
			job.Metadata = coremodel.JSONB{}
		}
		guardMetadata := coremodel.JSONB{
			"slowdown_source":     "coordination_guard",
			"coordination_state":  "account_overlap",
			"runtime_guard_scope": "in_process_lock",
		}
		for key, value := range guardMetadata {
			job.Metadata[key] = value
		}
		_ = s.jobs.UpdateFields(ctx, job.ID, map[string]interface{}{
			"status":        job.Status,
			"error_message": job.ErrorMessage,
			"metadata":      job.Metadata,
		})
		s.recordJobEvent(ctx, job.ID, "warning", "Skipped sync; account already syncing", map[string]interface{}{
			"account_id":          account.ID,
			"slowdown_source":     "coordination_guard",
			"coordination_state":  "account_overlap",
			"runtime_guard_scope": "in_process_lock",
		})
		return
	}
	defer func() {
		if locked {
			s.unlockAccount(account.ID)
		}
	}()

	// Update job status to running (use UpdateFields to preserve metadata)
	job.Status = caasmmodel.SyncJobRunning
	now := time.Now()
	job.StartedAt = &now
	_ = s.jobs.UpdateFields(ctx, job.ID, map[string]interface{}{
		"status":     caasmmodel.SyncJobRunning,
		"started_at": now,
	})

	// Initialize Event Reporter
	var reporter *DBEventReporter
	if s.events != nil {
		reporter = &DBEventReporter{
			repo:      s.events,
			syncJobID: job.ID,
			ctx:       ctx,
			logger:    s.logger,
		}
	}

	var err error
	hadPartialFailures := false
	partialFailureCount := 0
	allowStaleMark := false
	var runStart time.Time
	persistCtx := context.Background()
	defer func() {
		finishedAt := time.Now()
		job.FinishedAt = &finishedAt
		var killSwitchErr *KillSwitchError
		blockedByKillSwitch := err != nil && errors.As(err, &killSwitchErr)
		externalStop := s.wasJobStoppedExternally(job.ID)
		contextSource, _ := job.Metadata["context_source"].(string)
		executionTimeoutMinutes, _ := job.Metadata["execution_timeout_minutes"].(int)
		if executionTimeoutMinutes <= 0 {
			switch value := job.Metadata["execution_timeout_minutes"].(type) {
			case int64:
				executionTimeoutMinutes = int(value)
			case float64:
				executionTimeoutMinutes = int(value)
			}
		}
		if executionTimeoutMinutes <= 0 {
			executionTimeoutMinutes = s.resolveExecutionTimeoutMinutes()
		}
		failedPhase, _ := job.Metadata["failed_phase"].(string)
		if externalStop {
			allowStaleMark = false
			job.Status = caasmmodel.SyncJobFailed
			job.ErrorMessage = "Job stopped by user"
			s.updateSyncJobMetadata(persistCtx, job, coremodel.JSONB{
				"termination_reason":        "job_stopped",
				"cancellation_source":       "user_stop",
				"stop_requested":            true,
				"stale_mark_completed":      false,
				"full_run_completed":        false,
				"stale_mark_eligible":       false,
				"stale_mark_skipped_reason": "job_stopped",
			})
			if reporter != nil {
				reporter.Report("warning", job.ErrorMessage, map[string]interface{}{
					"stop_state": "external_stop",
				})
			}
		} else if errors.Is(ctx.Err(), context.DeadlineExceeded) {
			allowStaleMark = false
			job.Status = caasmmodel.SyncJobFailed
			job.ErrorMessage = fmt.Sprintf("Job timed out (execution exceeded %d minutes)", executionTimeoutMinutes)
			s.updateSyncJobMetadata(persistCtx, job, coremodel.JSONB{
				"termination_reason":        "execution_timeout",
				"cancellation_source":       "execution_timeout",
				"stale_mark_skipped_reason": "execution_timeout",
			})
			if reporter != nil && !blockedByKillSwitch {
				reporter.Report("error", job.ErrorMessage, map[string]interface{}{
					"error":                     job.ErrorMessage,
					"failed_phase":              job.Metadata["failed_phase"],
					"termination_reason":        "execution_timeout",
					"cancellation_source":       "execution_timeout",
					"context_source":            contextSource,
					"execution_timeout_minutes": executionTimeoutMinutes,
				})
			}
			if !blockedByKillSwitch {
				s.setBackoff(account.ID, 2*time.Minute)
			}
		} else if errors.Is(ctx.Err(), context.Canceled) {
			allowStaleMark = false
			cancellationSource := deriveCancellationSource(ctx, contextSource)
			job.Status = caasmmodel.SyncJobFailed
			job.ErrorMessage = "Job interrupted by external cancellation"
			s.updateSyncJobMetadata(persistCtx, job, coremodel.JSONB{
				"termination_reason":        "external_cancellation",
				"cancellation_source":       cancellationSource,
				"stale_mark_skipped_reason": "sync_error",
			})
			if reporter != nil && !blockedByKillSwitch {
				reporter.Report("error", job.ErrorMessage, map[string]interface{}{
					"error":                 errString(err, job.ErrorMessage),
					"failed_phase":          job.Metadata["failed_phase"],
					"termination_reason":    "external_cancellation",
					"cancellation_source":   cancellationSource,
					"context_source":        contextSource,
					"partial_failure_count": job.Metadata["partial_failure_count"],
					"partial_failures":      job.Metadata["partial_failures"],
				})
			}
			if !blockedByKillSwitch {
				s.setBackoff(account.ID, 2*time.Minute)
			}
		} else if err != nil && failedPhase == "persist" && isPersistContextError(err) {
			allowStaleMark = false
			cancellationSource, _ := job.Metadata["cancellation_source"].(string)
			if strings.TrimSpace(cancellationSource) == "" {
				cancellationSource = derivePersistCancellationSource(err)
			}
			job.Status = caasmmodel.SyncJobFailed
			if cancellationSource == "db_query_deadline_exceeded" {
				job.ErrorMessage = "Persist stage database query timed out before snapshot completion"
			} else {
				job.ErrorMessage = "Persist stage database query was canceled before snapshot completion"
			}
			s.updateSyncJobMetadata(persistCtx, job, coremodel.JSONB{
				"termination_reason":        "persist_error",
				"cancellation_source":       cancellationSource,
				"stale_mark_skipped_reason": "persist_error",
			})
			if reporter != nil && !blockedByKillSwitch {
				reporter.Report("error", job.ErrorMessage, map[string]interface{}{
					"error":                            err.Error(),
					"failed_phase":                     failedPhase,
					"termination_reason":               "persist_error",
					"cancellation_source":              cancellationSource,
					"context_source":                   contextSource,
					"persist_asset_count":              job.Metadata["persist_asset_count"],
					"persist_prefetch_native_id_count": job.Metadata["persist_prefetch_native_id_count"],
					"persist_prefetch_chunk_count":     job.Metadata["persist_prefetch_chunk_count"],
					"persist_prefetch_hit_count":       job.Metadata["persist_prefetch_hit_count"],
					"persist_prefetch_miss_count":      job.Metadata["persist_prefetch_miss_count"],
					"partial_failure_count":            job.Metadata["partial_failure_count"],
					"partial_failures":                 job.Metadata["partial_failures"],
				})
			}
			if !blockedByKillSwitch {
				s.setBackoff(account.ID, 2*time.Minute)
			}
		} else if err != nil {
			job.Status = caasmmodel.SyncJobFailed
			job.ErrorMessage = err.Error()
			s.updateSyncJobMetadata(persistCtx, job, coremodel.JSONB{
				"termination_reason":        "sync_error",
				"stale_mark_skipped_reason": "sync_error",
			})
			if reporter != nil && !blockedByKillSwitch {
				reporter.Report("error", fmt.Sprintf("Job failed: %v", err), map[string]interface{}{
					"error":                 err.Error(),
					"failed_phase":          job.Metadata["failed_phase"],
					"termination_reason":    "sync_error",
					"context_source":        contextSource,
					"partial_failure_count": job.Metadata["partial_failure_count"],
					"partial_failures":      job.Metadata["partial_failures"],
				})
			}
			if !blockedByKillSwitch {
				s.setBackoff(account.ID, 2*time.Minute)
			}
		} else if hadPartialFailures {
			job.Status = caasmmodel.SyncJobPartial
			job.ErrorMessage = fmt.Sprintf("sync completed with %d partial failures", partialFailureCount)
			s.setBackoff(account.ID, 2*time.Minute)
			s.updateSyncJobMetadata(persistCtx, job, coremodel.JSONB{
				"termination_reason":        "partial_failures",
				"stale_mark_skipped_reason": "partial_failures",
			})
			if reporter != nil {
				reporter.Report("warning", job.ErrorMessage, map[string]interface{}{
					"partial_failure_count": job.Metadata["partial_failure_count"],
					"partial_failures":      job.Metadata["partial_failures"],
				})
			}
		} else {
			job.Status = caasmmodel.SyncJobSuccess
			s.clearBackoff(account.ID)
			if reporter != nil {
				reporter.Report("info", "Job completed successfully", nil)
			}
		}
		if allowStaleMark && !runStart.IsZero() {
			s.finalizeStaleMarking(persistCtx, job, account.ID, runStart, true)
		}
		// Use UpdateFields to preserve metadata set by TriggerSyncWithPriority
		updateFields := map[string]interface{}{
			"status":      job.Status,
			"finished_at": finishedAt,
		}
		if job.ErrorMessage != "" {
			updateFields["error_message"] = job.ErrorMessage
		}
		terminalPersisted := true
		switch job.Status {
		case caasmmodel.SyncJobSuccess:
			terminalPersisted = s.persistSyncJobSuccessIfActive(persistCtx, job, finishedAt)
		case caasmmodel.SyncJobFailed:
			terminalPersisted = s.persistSyncJobFailureIfActive(persistCtx, job, finishedAt)
		case caasmmodel.SyncJobPartial:
			terminalPersisted = s.persistSyncJobTerminalIfActive(persistCtx, job, updateFields)
		default:
			if s.jobs != nil {
				_ = s.jobs.UpdateFields(persistCtx, job.ID, updateFields)
			}
		}

		// Record last sync time and error on the cloud account so the scheduler respects intervals.
		if s.accounts != nil && !blockedByKillSwitch && terminalPersisted {
			if job.Status == caasmmodel.SyncJobSuccess {
				account.LastSyncAt = &finishedAt
				account.ErrorMessage = ""
			} else {
				if err != nil {
					account.ErrorMessage = err.Error()
				} else {
					account.ErrorMessage = job.ErrorMessage
				}
			}
			_ = s.accounts.Update(persistCtx, account)
		}
	}()

	// Initialize Collector
	var col collector.Collector
	var region string

	if err = s.blockIfKillSwitchEnabled(ctx, job, account.ID, "before_collector_init"); err != nil {
		return
	}
	col, region, err = s.buildCollector(ctx, account)
	if err != nil {
		var identityErr *awsIdentityValidationError
		if errors.As(err, &identityErr) {
			s.recordAWSIdentityValidationFailure(ctx, job, identityErr)
		}
		return
	}

	windowStartSeed := time.Now()
	resumeCursor := resumeCursorFromMetadata(job.Metadata)
	var resumeSourceJobID uint64
	if !resumeCursorHasProgress(resumeCursor) {
		resumeCursor, resumeSourceJobID = loadResumeCursor(ctx, s.jobs, account.ID, job.ID, job.JobType)
	}
	resumeCursor = normalizeResumeCursor(resumeCursor, windowStartSeed)
	runStart = resumeWindowStart(resumeCursor, windowStartSeed)
	if setter, ok := col.(collectorResumeSetter); ok {
		cursorCopy := resumeCursor
		setter.SetResumeCursor(&cursorCopy)
	}
	s.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
		"collection_completed":  false,
		"persist_completed":     false,
		"stale_mark_completed":  false,
		"stale_mark_error":      "",
		"full_run_completed":    false,
		"stale_mark_eligible":   false,
		"resume_cursor":         resumeCursorJSONB(resumeCursor),
		"resume_window_started": runStart.Format(time.RFC3339Nano),
	})
	if resumeSourceJobID != 0 {
		s.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
			"resume_source_job_id": resumeSourceJobID,
		})
	}

	// Set Event Reporter on Collector
	if reporter != nil {
		col.SetEventReporter(reporter)
		reporter.Report("info", "Starting collection", map[string]interface{}{
			"provider": account.Provider,
			"region":   region,
		})
	}
	if err = s.blockIfKillSwitchEnabled(ctx, job, account.ID, "before_collect"); err != nil {
		return
	}

	if chunkCollector, ok := col.(collector.ChunkCollector); ok {
		sink := newChunkedSyncSink(s, job, account, ctx, resumeCursor, resumeSourceJobID, runStart)
		summary, collectErr := chunkCollector.CollectChunks(ctx, sink)
		waitErr := sink.Wait()
		if collectErr != nil {
			if summary == nil {
				summary = &collector.CollectionSummary{}
			}
			if waitErr != nil {
				s.recordJobEvent(ctx, job.ID, "warning", "Chunk persist writer also failed after collection error", map[string]interface{}{
					"collect_error": collectErr.Error(),
					"writer_error":  waitErr.Error(),
				})
			}
			if shouldTreatCollectionErrorAsPartial(collectErr) {
				summary.PartialFailures = append(summary.PartialFailures, collector.CollectionIssue{
					Scope:   "collection",
					Message: collectErr.Error(),
				})
				applyPartialFailureMetadata(job, summary.PartialFailures)
				s.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
					"failed_phase": "collection",
				})
				s.recordJobEvent(ctx, job.ID, "warning", "Collection degraded to partial failure", map[string]interface{}{
					"error":                 collectErr.Error(),
					"failed_phase":          "collection",
					"partial_failure_count": len(summary.PartialFailures),
					"partial_failures":      summary.PartialFailures,
				})
			} else {
				applyPartialFailureMetadata(job, summary.PartialFailures)
				s.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
					"failed_phase": "collection",
				})
				s.recordJobEvent(ctx, job.ID, "error", "Collection failed", map[string]interface{}{
					"error":                 collectErr.Error(),
					"failed_phase":          "collection",
					"partial_failure_count": len(summary.PartialFailures),
					"partial_failures":      summary.PartialFailures,
				})
				err = collectErr
				return
			}
		}
		if waitErr != nil {
			if summary == nil {
				summary = &collector.CollectionSummary{}
			}
			if shouldTreatChunkPersistErrorAsPartial(waitErr) {
				summary.PartialFailures = append(summary.PartialFailures, collector.CollectionIssue{
					Scope:   "chunk_persist",
					Message: waitErr.Error(),
				})
			} else {
				err = waitErr
				return
			}
		}
		if summary == nil {
			summary = &collector.CollectionSummary{}
		}
		if controlMessage, hasControlIssue := partialFailureControlMessage(summary.PartialFailures); hasControlIssue {
			s.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
				"failed_phase": "collection",
			})
			s.recordJobEvent(ctx, job.ID, "error", "Collection interrupted by control signal", map[string]interface{}{
				"error":                 controlMessage,
				"failed_phase":          "collection",
				"partial_failure_count": len(summary.PartialFailures),
				"partial_failures":      summary.PartialFailures,
			})
			err = errors.New(controlMessage)
			return
		}
		s.recordJobEvent(ctx, job.ID, "info", chunkCollectionMessage(summary, sink), map[string]interface{}{
			"chunk_count":             summary.ChunkCount,
			"persisted_chunk_count":   sink.chunkCount,
			"persist_queue_capacity":  chunkPersistQueueCapacity,
			"persist_queue_depth_max": sink.queueDepthMax,
			"asset_count":             sink.assetCount,
			"ip_asset_count":          sink.ipAssetCount,
			"domain_asset_count":      sink.domainCount,
			"port_count":              sink.portCount,
			"partial_failure_count":   len(summary.PartialFailures),
		})
		fullRunCompleted := len(summary.PartialFailures) == 0
		s.updateSyncJobMetadata(ctx, job, chunkSummaryMetadata(summary, sink, fullRunCompleted))
		if len(summary.PartialFailures) > 0 {
			hadPartialFailures = true
			partialFailureCount = len(summary.PartialFailures)
			applyPartialFailureMetadata(job, summary.PartialFailures)
			if updateErr := s.jobs.UpdateFields(ctx, job.ID, map[string]interface{}{
				"metadata": job.Metadata,
			}); updateErr != nil {
				s.logError("failed to persist partial failure metadata",
					"job_id", job.ID,
					"cloud_account_id", account.ID,
					"phase", "collect_chunks",
					"error", updateErr,
				)
			}
		}
		allowStaleMark = fullRunCompleted
		return
	}

	// Collect
	result, err := col.Collect(ctx)
	if err != nil {
		s.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
			"failed_phase": "collection",
		})
		if shouldTreatCollectionErrorAsPartial(err) {
			hadPartialFailures = true
			partialFailureCount = 1
			issues := []collector.CollectionIssue{{
				Scope:   "collection",
				Message: err.Error(),
			}}
			applyPartialFailureMetadata(job, issues)
			s.recordJobEvent(ctx, job.ID, "warning", "Collection degraded to partial failure", map[string]interface{}{
				"error":                 err.Error(),
				"failed_phase":          "collection",
				"partial_failure_count": 1,
				"partial_failures":      issues,
			})
			err = nil
			return
		}
		s.recordJobEvent(ctx, job.ID, "error", "Collection failed", map[string]interface{}{
			"error":        err.Error(),
			"failed_phase": "collection",
		})
		return
	}
	s.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
		"collection_completed": true,
	})
	s.recordJobEvent(ctx, job.ID, "info", "Collection finished", map[string]interface{}{
		"asset_count":        len(result.Assets),
		"ip_asset_count":     len(result.IPAssets),
		"domain_asset_count": len(result.DomainAssets),
		"port_count":         len(result.Ports),
	})
	if controlMessage, hasControlIssue := partialFailureControlMessage(result.PartialFailures); hasControlIssue {
		s.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
			"failed_phase": "collection",
		})
		s.recordJobEvent(ctx, job.ID, "error", "Collection interrupted by control signal", map[string]interface{}{
			"error":                 controlMessage,
			"failed_phase":          "collection",
			"partial_failure_count": len(result.PartialFailures),
			"partial_failures":      result.PartialFailures,
		})
		err = errors.New(controlMessage)
		return
	}

	if len(result.PartialFailures) > 0 {
		hadPartialFailures = true
		partialFailureCount = len(result.PartialFailures)
		applyPartialFailureMetadata(job, result.PartialFailures)
		s.recordJobEvent(ctx, job.ID, "warning", fmt.Sprintf("Collection completed with %d partial failures", len(result.PartialFailures)), map[string]interface{}{
			"partial_failure_count": len(result.PartialFailures),
			"partial_failures":      result.PartialFailures,
		})
		if updateErr := s.jobs.UpdateFields(ctx, job.ID, map[string]interface{}{
			"metadata": job.Metadata,
		}); updateErr != nil {
			s.logError("failed to persist partial failure metadata",
				"job_id", job.ID,
				"cloud_account_id", account.ID,
				"phase", "collect",
				"error", updateErr,
			)
		}
	}

	payload := &caasmmodel.IngestPayload{
		SyncJobID:      job.ID,
		CloudAccountID: account.ID,
		Assets:         result.Assets,
		IPAssets:       result.IPAssets,
		DomainAssets:   result.DomainAssets,
		Ports:          result.Ports,
	}
	if err = s.blockIfKillSwitchEnabled(ctx, job, account.ID, "before_persist"); err != nil {
		return
	}
	s.filterSyncPayloadByCollectionScope(ctx, payload)
	if pruneErr := s.prunePersistedDataByCollectionScope(ctx, account.ID); pruneErr != nil {
		err = pruneErr
		return
	}
	if persistErr := s.persistSyncPayload(ctx, job, payload); persistErr != nil {
		err = persistErr
		return
	}
	s.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
		"persist_completed": true,
	})
	s.recordJobEvent(ctx, job.ID, "info", "Persist finished", map[string]interface{}{
		"asset_count":        len(payload.Assets),
		"ip_asset_count":     len(payload.IPAssets),
		"domain_asset_count": len(payload.DomainAssets),
		"port_count":         len(payload.Ports),
	})
	s.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
		"full_run_completed":  !hadPartialFailures,
		"stale_mark_eligible": !hadPartialFailures,
	})
	allowStaleMark = !hadPartialFailures

}

func shouldTreatChunkPersistErrorAsPartial(err error) bool {
	_ = err
	return false
}

func shouldTreatCollectionErrorAsPartial(err error) bool {
	if err == nil {
		return false
	}
	var killSwitchErr *KillSwitchError
	return !errors.As(err, &killSwitchErr) &&
		!errors.Is(err, context.Canceled) &&
		!errors.Is(err, context.DeadlineExceeded)
}

func (s *CloudAccountService) finalizeStaleMarking(ctx context.Context, job *caasmmodel.SyncJob, accountID uint64, runStart time.Time, shouldMarkStale bool) {
	if !shouldMarkStale {
		return
	}
	if err := s.softDeleteStaleAssets(ctx, accountID, runStart); err != nil {
		s.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
			"stale_mark_error": err.Error(),
		})
		s.recordJobEvent(ctx, job.ID, "warning", "Stale marking failed", map[string]interface{}{
			"cloud_account_id": accountID,
			"error":            err.Error(),
		})
		return
	}
	s.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
		"stale_mark_completed": true,
	})
	s.recordJobEvent(ctx, job.ID, "info", "Stale marking finished", map[string]interface{}{
		"cloud_account_id": accountID,
	})
}

func (s *CloudAccountService) wasJobStoppedExternally(jobID uint64) bool {
	if s.jobs == nil || jobID == 0 {
		return false
	}
	job, err := s.jobs.FindByID(context.Background(), jobID)
	if err != nil || job == nil {
		return false
	}
	if job.Metadata != nil {
		switch value := job.Metadata["stop_requested"].(type) {
		case bool:
			if value {
				return true
			}
		case string:
			if strings.EqualFold(strings.TrimSpace(value), "true") {
				return true
			}
		}
	}
	return job.Status == caasmmodel.SyncJobFailed && strings.TrimSpace(job.ErrorMessage) == "Job stopped by user"
}

func syncJobDurableContext(ctx context.Context) context.Context {
	if ctx == nil || ctx.Err() != nil {
		return context.Background()
	}
	return ctx
}

func isSyncJobTerminalStatus(status caasmmodel.SyncJobStatus) bool {
	return status == caasmmodel.SyncJobSuccess ||
		status == caasmmodel.SyncJobPartial ||
		status == caasmmodel.SyncJobFailed
}

func isSyncJobStartableStatus(status caasmmodel.SyncJobStatus) bool {
	return status == caasmmodel.SyncJobQueued || status == caasmmodel.SyncJobPending
}

func (s *CloudAccountService) persistSyncJobSuccessIfActive(ctx context.Context, job *caasmmodel.SyncJob, finishedAt time.Time) bool {
	return s.persistSyncJobTerminalIfActive(ctx, job, map[string]interface{}{
		"status":      caasmmodel.SyncJobSuccess,
		"finished_at": finishedAt,
	})
}

func (s *CloudAccountService) persistSyncJobFailureIfActive(ctx context.Context, job *caasmmodel.SyncJob, finishedAt time.Time) bool {
	if job == nil {
		return false
	}
	return s.persistSyncJobTerminalIfActive(ctx, job, map[string]interface{}{
		"status":        caasmmodel.SyncJobFailed,
		"error_message": job.ErrorMessage,
		"finished_at":   finishedAt,
	})
}

func (s *CloudAccountService) persistSyncJobTerminalIfActive(ctx context.Context, job *caasmmodel.SyncJob, fields map[string]interface{}) bool {
	if s.jobs == nil || job == nil || job.ID == 0 {
		return false
	}
	writeCtx := syncJobDurableContext(ctx)
	latest, err := s.jobs.FindByID(writeCtx, job.ID)
	if err != nil || latest == nil || (latest != job && isSyncJobTerminalStatus(latest.Status)) {
		return false
	}
	return s.jobs.UpdateFields(writeCtx, job.ID, fields) == nil
}

func (s *CloudAccountService) recordAWSIdentityValidationFailure(ctx context.Context, job *caasmmodel.SyncJob, validationErr *awsIdentityValidationError) {
	if job == nil || validationErr == nil {
		return
	}
	if job.Metadata == nil {
		job.Metadata = coremodel.JSONB{}
	}
	job.Metadata["aws_identity_validation_failed"] = true
	job.Metadata["aws_identity_expected_account_id"] = validationErr.ExpectedAccountID
	job.Metadata["aws_identity_role_account_id"] = validationErr.RoleAccountID
	job.Metadata["aws_identity_actual_account_id"] = validationErr.ActualAccountID
	job.Metadata["aws_identity_actual_arn"] = validationErr.ActualARN
	if s.jobs != nil && job.ID != 0 {
		_ = s.jobs.UpdateFields(syncJobDurableContext(ctx), job.ID, map[string]interface{}{
			"metadata": job.Metadata,
		})
	}
	s.recordJobEvent(syncJobDurableContext(ctx), job.ID, "error", "AWS identity validation failed", map[string]interface{}{
		"expected_account_id": validationErr.ExpectedAccountID,
		"role_account_id":     validationErr.RoleAccountID,
		"actual_account_id":   validationErr.ActualAccountID,
		"actual_arn":          validationErr.ActualARN,
	})
}

func (s *CloudAccountService) persistPayloadAndMarkStale(ctx context.Context, job *caasmmodel.SyncJob, account *caasmmodel.CloudAccount, payload *caasmmodel.IngestPayload, runStart time.Time) error {
	if payload == nil {
		return errors.New("empty sync payload")
	}
	if account != nil && payload.CloudAccountID == 0 {
		payload.CloudAccountID = account.ID
	}
	s.filterSyncPayloadByCollectionScope(ctx, payload)
	if account != nil {
		if err := s.prunePersistedDataByCollectionScope(ctx, account.ID); err != nil {
			return err
		}
	}
	if err := s.persistSyncPayload(ctx, job, payload); err != nil {
		return err
	}
	if job != nil {
		s.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
			"persist_completed": true,
		})
		s.recordJobEvent(ctx, job.ID, "info", "Persist finished", map[string]interface{}{
			"asset_count":        len(payload.Assets),
			"ip_asset_count":     len(payload.IPAssets),
			"domain_asset_count": len(payload.DomainAssets),
			"port_count":         len(payload.Ports),
		})
	}
	if account != nil && !runStart.IsZero() {
		s.finalizeStaleMarking(ctx, job, account.ID, runStart, true)
	}
	return nil
}

func (s *CloudAccountService) processCollectedResult(ctx context.Context, job *caasmmodel.SyncJob, account *caasmmodel.CloudAccount, result *collector.AssetResult, runStart time.Time) error {
	if result == nil {
		return errors.New("empty collection result")
	}
	if controlMessage, hasControlIssue := partialFailureControlMessage(result.PartialFailures); hasControlIssue {
		if job != nil {
			s.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
				"failed_phase": "collection",
			})
			s.recordJobEvent(ctx, job.ID, "error", "Collection interrupted by control signal", map[string]interface{}{
				"error":                 controlMessage,
				"failed_phase":          "collection",
				"partial_failure_count": len(result.PartialFailures),
				"partial_failures":      result.PartialFailures,
			})
		}
		return errors.New(controlMessage)
	}
	if len(result.PartialFailures) > 0 {
		return s.handleCollectionPartialFailures(ctx, job, result)
	}
	cloudAccountID := uint64(0)
	if account != nil {
		cloudAccountID = account.ID
	}
	syncJobID := uint64(0)
	if job != nil {
		syncJobID = job.ID
	}
	return s.persistPayloadAndMarkStale(ctx, job, account, &caasmmodel.IngestPayload{
		SyncJobID:      syncJobID,
		CloudAccountID: cloudAccountID,
		Assets:         result.Assets,
		IPAssets:       result.IPAssets,
		DomainAssets:   result.DomainAssets,
		Ports:          result.Ports,
	}, runStart)
}

func (s *CloudAccountService) softDeleteStaleAssets(ctx context.Context, accountID uint64, runStart time.Time) error {
	if s.assets == nil {
		return nil
	}
	if err := s.assets.SoftDeleteStale(ctx, accountID, runStart); err != nil {
		s.logError("failed to soft-delete stale CAASM assets",
			"cloud_account_id", accountID,
			"phase", "stale_mark",
			"error", err,
		)
		return err
	}
	return nil
}

func (s *CloudAccountService) updateSyncJobMetadata(ctx context.Context, job *caasmmodel.SyncJob, values coremodel.JSONB) error {
	if s.jobs == nil || job == nil || job.ID == 0 || len(values) == 0 {
		return nil
	}
	nextMetadata := coremodel.JSONB{}
	for key, value := range job.Metadata {
		nextMetadata[key] = value
	}
	for key, value := range values {
		nextMetadata[key] = value
	}
	if err := s.jobs.UpdateFields(ctx, job.ID, map[string]interface{}{
		"metadata": nextMetadata,
	}); err != nil {
		s.logError("failed to update sync job metadata",
			"job_id", job.ID,
			"phase", "metadata_update",
			"error", err,
		)
		return err
	}
	job.Metadata = nextMetadata
	return nil
}

func summarizeThrottleIssue(issues []collector.CollectionIssue) (bool, string) {
	for _, issue := range issues {
		signal := strings.ToLower(strings.TrimSpace(issue.Message + " " + issue.Target + " " + issue.Scope))
		if strings.Contains(signal, "throttl") || strings.Contains(signal, "rate exceeded") || strings.Contains(signal, "too many request") {
			target := strings.TrimSpace(issue.Target)
			if target == "" {
				target = strings.TrimSpace(issue.Scope)
			}
			return true, target
		}
	}
	return false, ""
}

func applyPartialFailureMetadata(job *caasmmodel.SyncJob, issues []collector.CollectionIssue) {
	if job == nil || len(issues) == 0 {
		return
	}
	if job.Metadata == nil {
		job.Metadata = coremodel.JSONB{}
	}
	job.Metadata["partial_failure_count"] = len(issues)
	job.Metadata["partial_failures"] = issues
	job.Metadata["completed_with_warnings"] = true
	job.Metadata["stale_mark_skipped_reason"] = "partial_failures"
	if hasThrottle, throttleTarget := summarizeThrottleIssue(issues); hasThrottle {
		job.Metadata["slowdown_source"] = "aws_throttle"
		job.Metadata["throttle_state"] = "partial_failure"
		if throttleTarget != "" {
			job.Metadata["throttle_target"] = throttleTarget
		}
	}
}

func partialFailureControlMessage(issues []collector.CollectionIssue) (string, bool) {
	for _, issue := range issues {
		message := strings.TrimSpace(issue.Message)
		lower := strings.ToLower(strings.TrimSpace(issue.Message + " " + issue.Target + " " + issue.Scope))
		switch {
		case strings.Contains(lower, "context canceled"):
			return "context canceled", true
		case strings.Contains(lower, "deadline exceeded"):
			return "context deadline exceeded", true
		case strings.Contains(lower, strings.ToLower(defaultKillSwitchMessage)):
			if message == "" {
				message = defaultKillSwitchMessage
			}
			return message, true
		}
	}
	return "", false
}

type collectionPartialFailureError struct {
	Issues []collector.CollectionIssue
}

func (e *collectionPartialFailureError) Error() string {
	return fmt.Sprintf("collection failed with %d partial failures", len(e.Issues))
}

func (s *CloudAccountService) handleCollectionPartialFailures(ctx context.Context, job *caasmmodel.SyncJob, result *collector.AssetResult) error {
	if result == nil || len(result.PartialFailures) == 0 {
		return nil
	}
	applyPartialFailureMetadata(job, result.PartialFailures)
	if job != nil {
		if job.Metadata == nil {
			job.Metadata = coremodel.JSONB{}
		}
		job.Metadata["collection_incomplete"] = true
		job.Metadata["stale_marking_skipped"] = true
		job.Metadata["stale_mark_eligible"] = false
		job.Metadata["stale_mark_completed"] = false
	}

	eventMetadata := map[string]interface{}{
		"failed_phase":          "collection",
		"partial_failure_count": len(result.PartialFailures),
		"partial_failures":      result.PartialFailures,
		"stale_marking_skipped": true,
	}
	var updateErr error
	if s.jobs != nil && job != nil && job.ID != 0 {
		updateErr = s.jobs.UpdateFields(syncJobDurableContext(ctx), job.ID, map[string]interface{}{
			"metadata": job.Metadata,
		})
		if updateErr != nil {
			eventMetadata["metadata_update_error"] = updateErr.Error()
		}
	}
	if job != nil {
		s.recordJobEvent(syncJobDurableContext(ctx), job.ID, "error", "Collection failed with partial failures", eventMetadata)
	}
	if updateErr != nil {
		return fmt.Errorf("persist partial failure metadata: %w", updateErr)
	}
	return &collectionPartialFailureError{Issues: result.PartialFailures}
}

func (s *CloudAccountService) recordHistory(ctx context.Context, assetID uint64, action string, changes coremodel.JSONB) error {
	if s.history == nil || assetID == 0 {
		return nil
	}
	if changes == nil {
		changes = coremodel.JSONB{}
	}
	entry := &caasmmodel.AssetHistory{
		AssetID:   assetID,
		Action:    action,
		ChangedAt: time.Now(),
		ChangedBy: "collector",
		Changes:   changes,
	}
	return s.history.Create(ctx, entry)
}

func (s *CloudAccountService) recordJobEvent(ctx context.Context, jobID uint64, eventType string, message string, metadata map[string]interface{}) {
	if s.events == nil || jobID == 0 {
		return
	}
	if metadata == nil {
		metadata = map[string]interface{}{}
	}
	s.persistCoordinationGuardMetadata(ctx, jobID, metadata)
	if err := s.events.Create(ctx, &caasmmodel.SyncJobEvent{
		SyncJobID: jobID,
		Type:      caasmmodel.SyncJobEventType(eventType),
		Message:   message,
		Metadata:  coremodel.JSONB(metadata),
	}); err != nil {
		s.logError("failed to record CAASM sync job event",
			"job_id", jobID,
			"event_type", eventType,
			"message", message,
			"phase", metadata["phase"],
			"failed_phase", metadata["failed_phase"],
			"account_id", metadata["account_id"],
			"cloud_account_id", metadata["cloud_account_id"],
			"error", err,
		)
	}
}

func (s *CloudAccountService) persistCoordinationGuardMetadata(ctx context.Context, jobID uint64, metadata map[string]interface{}) {
	if s.jobs == nil || len(metadata) == 0 {
		return
	}
	if !hasCoordinationGuardMetadata(metadata) {
		return
	}
	job, err := s.jobs.FindByID(ctx, jobID)
	if err != nil || job == nil {
		return
	}
	if job.Metadata == nil {
		job.Metadata = coremodel.JSONB{}
	}
	for _, key := range []string{
		"slowdown_source",
		"coordination_state",
		"runtime_guard_scope",
		"reason",
		"retry_after_sec",
		"consumer",
	} {
		if value, ok := metadata[key]; ok {
			job.Metadata[key] = value
		}
	}
	if err := s.jobs.UpdateFields(ctx, job.ID, map[string]interface{}{
		"metadata": job.Metadata,
	}); err != nil {
		s.logError("failed to persist coordination guard metadata",
			"job_id", job.ID,
			"phase", "coordination_guard",
			"error", err,
		)
	}
}

func hasCoordinationGuardMetadata(metadata map[string]interface{}) bool {
	if metadata == nil {
		return false
	}
	if source, ok := metadata["slowdown_source"].(string); ok && source == "coordination_guard" {
		return true
	}
	if _, ok := metadata["coordination_state"]; ok {
		return true
	}
	if _, ok := metadata["runtime_guard_scope"]; ok {
		return true
	}
	return false
}

func extractTags(meta coremodel.JSONB) (map[string]string, bool) {
	raw, ok := meta["tags"]
	if !ok {
		return nil, false
	}
	result := make(map[string]string)
	switch t := raw.(type) {
	case map[string]interface{}:
		for k, v := range t {
			if vs, ok := v.(string); ok {
				result[k] = vs
			}
		}
	case map[string]string:
		for k, v := range t {
			result[k] = v
		}
	}
	if len(result) == 0 {
		return nil, false
	}
	return result, true
}

func (s *CloudAccountService) invokeLambdaSimulated(ctx context.Context, functionName string, payload *caasmqueue.SyncJobPayload) error {
	runCtx := ctx
	if payload != nil && payload.SyncJobID != nil && *payload.SyncJobID != 0 {
		var cancel context.CancelFunc
		runCtx, cancel = context.WithCancel(ctx)
		s.registerSyncCancel(*payload.SyncJobID, cancel)
		defer func() {
			cancel()
			s.clearSyncCancel(*payload.SyncJobID)
		}()
	}

	if payload != nil && payload.SyncJobID != nil {
		s.recordJobEvent(runCtx, *payload.SyncJobID, "info", "Serverless worker execution started", map[string]interface{}{
			"worker_label": functionName,
			"mode":         "serverless",
			"priority":     payload.Priority,
		})
	}

	// This is the current runtime path for mode=serverless inside the API service.
	// It intentionally simulates asynchronous worker execution inline so the feature
	// remains testable and runnable without deploying an external AWS Lambda worker.
	// The repository also contains a real Lambda entrypoint under cmd/lambda/collector,
	// but this method does not invoke that binary or the AWS Lambda Invoke API.

	// For local testing, simulate the serverless worker by reusing the queue payload
	// execution path instead of calling runSync directly.
	err := s.ExecuteSyncPayload(runCtx, payload)
	if err != nil {
		if payload != nil && payload.SyncJobID != nil {
			s.recordJobEvent(runCtx, *payload.SyncJobID, "error", "Serverless worker execution failed", map[string]interface{}{
				"worker_label": functionName,
				"mode":         "serverless",
				"priority":     payload.Priority,
				"error":        err.Error(),
			})
		}
		return fmt.Errorf("ExecuteSyncPayload failed: %w", err)
	}

	if payload != nil && payload.SyncJobID != nil {
		s.recordJobEvent(runCtx, *payload.SyncJobID, "info", "Serverless worker execution completed", map[string]interface{}{
			"worker_label": functionName,
			"mode":         "serverless",
			"priority":     payload.Priority,
		})
	}
	return nil
}

type SyncJobService struct {
	jobs   repository.SyncJobRepository
	events repository.SyncJobEventRepository
}

type SyncJobListResult struct {
	Items []caasmmodel.SyncJob
	Total int64
}

func NewSyncJobService(repo repository.SyncJobRepository, eventRepo repository.SyncJobEventRepository) *SyncJobService {
	return &SyncJobService{jobs: repo, events: eventRepo}
}

func (s *SyncJobService) List(ctx context.Context, filter repository.SyncJobFilter, page, pageSize int) (*SyncJobListResult, error) {
	if s.jobs == nil {
		return &SyncJobListResult{}, nil
	}
	items, total, err := s.jobs.List(ctx, filter, page, pageSize)
	if err != nil {
		return nil, err
	}
	return &SyncJobListResult{Items: items, Total: total}, nil
}

func (s *SyncJobService) ListEvents(ctx context.Context, syncJobID uint64, opts repository.SyncJobEventListOptions) ([]caasmmodel.SyncJobEvent, error) {
	if s.events == nil {
		return nil, nil
	}
	events, err := s.events.List(ctx, syncJobID, opts)
	if err != nil {
		return nil, err
	}

	// Seed a synthetic event for legacy jobs that predate event logging
	if len(events) == 0 && s.jobs != nil && opts.BeforeID == nil && opts.AfterID == nil {
		job, jErr := s.jobs.FindByID(ctx, syncJobID)
		if jErr == nil && job != nil {
			synth := &caasmmodel.SyncJobEvent{
				SyncJobID: job.ID,
				Type:      caasmmodel.SyncJobEventType("info"),
				Message:   "Job created (legacy record - no detailed events)",
				Metadata: coremodel.JSONB{
					"status":   job.Status,
					"mode":     job.Metadata["execution_mode"],
					"provider": job.Metadata["provider"],
				},
			}
			_ = s.events.Create(ctx, synth)
			return []caasmmodel.SyncJobEvent{*synth}, nil
		}
	}

	return events, nil
}

func (s *SyncJobService) Get(ctx context.Context, id uint64) (*caasmmodel.SyncJob, error) {
	if s.jobs == nil {
		return nil, errors.New("sync job repository is not configured")
	}
	return s.jobs.FindByID(ctx, id)
}

func (s *DomainService) GetRegistrars(ctx context.Context) ([]string, error) {
	if s.domains == nil {
		return []string{}, nil
	}
	return s.domains.GetRegistrars(ctx)
}

func (s *SyncJobService) Stop(ctx context.Context, id uint64) error {
	if s.jobs == nil {
		return errors.New("sync job repository is not configured")
	}
	job, err := s.jobs.FindByID(ctx, id)
	if err != nil {
		return err
	}

	if job.Status == caasmmodel.SyncJobSuccess || job.Status == caasmmodel.SyncJobPartial || job.Status == caasmmodel.SyncJobFailed {
		return errors.New("job is already finished")
	}

	job.Status = caasmmodel.SyncJobFailed
	job.ErrorMessage = "Job stopped by user"
	finishedAt := time.Now()
	job.FinishedAt = &finishedAt
	if job.Metadata == nil {
		job.Metadata = coremodel.JSONB{}
	}
	job.Metadata["stop_requested"] = true
	job.Metadata["stale_mark_completed"] = false
	job.Metadata["full_run_completed"] = false
	job.Metadata["stale_mark_eligible"] = false
	job.Metadata["stale_mark_skipped_reason"] = "job_stopped"

	return s.jobs.Update(ctx, job)
}

// Start marks a sync job as running.
func (s *SyncJobService) Start(ctx context.Context, id uint64) error {
	if s.jobs == nil {
		return errors.New("sync job repository is not configured")
	}
	job, err := s.jobs.FindByID(ctx, id)
	if err != nil {
		return err
	}

	// Only allow transition from Queued or Pending.
	if !isSyncJobStartableStatus(job.Status) {
		// Idempotency: if already running, just return nil
		if job.Status == caasmmodel.SyncJobRunning {
			return nil
		}
		return fmt.Errorf("invalid state transition: cannot start job in state %s", job.Status)
	}

	job.Status = caasmmodel.SyncJobRunning
	now := time.Now()
	job.StartedAt = &now

	return s.jobs.UpdateFields(ctx, job.ID, map[string]interface{}{
		"status":     caasmmodel.SyncJobRunning,
		"started_at": now,
	})
}

// ReportEvent adds a new event to the sync job.
func (s *SyncJobService) ReportEvent(ctx context.Context, id uint64, eventType string, message string, metadata map[string]interface{}) error {
	if s.events == nil {
		return errors.New("event repository is not configured")
	}
	// Verify job exists (optional, but good for integrity)
	if _, err := s.jobs.FindByID(ctx, id); err != nil {
		return err
	}

	event := &caasmmodel.SyncJobEvent{
		SyncJobID: id,
		Type:      caasmmodel.SyncJobEventType(eventType),
		Message:   message,
		Metadata:  coremodel.JSONB(metadata),
	}
	return s.events.Create(ctx, event)
}

// Fail marks a sync job as failed with an error message.
func (s *SyncJobService) Fail(ctx context.Context, id uint64, errorMessage string) error {
	if s.jobs == nil {
		return errors.New("sync job repository is not configured")
	}
	job, err := s.jobs.FindByID(ctx, id)
	if err != nil {
		return err
	}

	// If already finished, don't overwrite (or maybe we should if it was success but actually failed?)
	// For now, assume terminal states are final.
	if job.Status == caasmmodel.SyncJobSuccess || job.Status == caasmmodel.SyncJobPartial || job.Status == caasmmodel.SyncJobFailed {
		return nil
	}

	job.Status = caasmmodel.SyncJobFailed
	job.ErrorMessage = errorMessage
	finishedAt := time.Now()
	job.FinishedAt = &finishedAt

	return s.jobs.UpdateFields(ctx, job.ID, map[string]interface{}{
		"status":        caasmmodel.SyncJobFailed,
		"error_message": errorMessage,
		"finished_at":   finishedAt,
	})
}

type DBEventReporter struct {
	repo      repository.SyncJobEventRepository
	syncJobID uint64
	ctx       context.Context
	logger    *logger.Logger
}

func (r *DBEventReporter) Report(eventType, message string, metadata map[string]interface{}) {
	if r.repo == nil {
		return
	}
	event := &caasmmodel.SyncJobEvent{
		SyncJobID: r.syncJobID,
		Type:      caasmmodel.SyncJobEventType(eventType),
		Message:   message,
		Metadata:  coremodel.JSONB(metadata),
	}
	if err := r.repo.Create(syncJobDurableContext(r.ctx), event); err != nil && r.logger != nil {
		r.logger.Error("failed to record CAASM sync job event",
			"job_id", r.syncJobID,
			"event_type", eventType,
			"message", message,
			"phase", metadata["phase"],
			"failed_phase", metadata["failed_phase"],
			"account_id", metadata["account_id"],
			"cloud_account_id", metadata["cloud_account_id"],
			"error", err,
		)
	}
}