package service

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

	coremodel "security-center-backend/internal/model"
	caasmmodel "security-center-backend/internal/modules/caasm/model"
	"security-center-backend/internal/modules/caasm/queue"
	"security-center-backend/internal/modules/caasm/repository"
)

type resolvedCollectionScope struct {
	IPScope     caasmmodel.CollectionScopeMode
	PortScope   caasmmodel.CollectionScopeMode
	DomainScope caasmmodel.CollectionScopeMode
}

type dnsRecordPersistIdentity struct {
	DomainAssetID    uint64
	NameKey          string
	TypeKey          string
	SetIdentifierKey string
	RegionKey        string
	FailoverKey      string
}

type retryableSyncJobError struct {
	SyncJobID uint64
	Status    caasmmodel.SyncJobStatus
	Reason    string
	Message   string
}

func (e *retryableSyncJobError) Error() string {
	if e == nil {
		return "retryable sync job failure"
	}
	if strings.TrimSpace(e.Message) != "" {
		return fmt.Sprintf("retryable sync job %d failed: %s", e.SyncJobID, e.Message)
	}
	if strings.TrimSpace(e.Reason) != "" {
		return fmt.Sprintf("retryable sync job %d failed: %s", e.SyncJobID, e.Reason)
	}
	return fmt.Sprintf("retryable sync job %d failed", e.SyncJobID)
}

func normalizePersistedDNSRecordOptionalValue(value string) string {
	value = strings.TrimSpace(value)
	if value == "" || strings.EqualFold(value, "<nil>") {
		return ""
	}
	return value
}

func normalizePersistedDNSRecordStringKey(value string, upper bool) string {
	value = normalizePersistedDNSRecordOptionalValue(value)
	if value == "" {
		return ""
	}
	if upper {
		return strings.ToUpper(value)
	}
	return strings.ToLower(value)
}

func normalizePersistedDNSRecordIdentity(record *caasmmodel.DNSRecord) (dnsRecordPersistIdentity, error) {
	if record == nil {
		return dnsRecordPersistIdentity{}, errors.New("dns record is nil")
	}
	if record.DomainAssetID == 0 {
		return dnsRecordPersistIdentity{}, repository.ErrDNSRecordDomainAssetIDRequired
	}

	nameKey := strings.TrimSuffix(strings.ToLower(strings.TrimSpace(record.Name)), ".")
	if nameKey == "" {
		return dnsRecordPersistIdentity{}, repository.ErrDNSRecordNameRequired
	}
	typeKey := strings.ToUpper(strings.TrimSpace(record.Type))
	if typeKey == "" {
		return dnsRecordPersistIdentity{}, repository.ErrDNSRecordTypeRequired
	}

	record.NameKey = nameKey
	record.TypeKey = typeKey
	record.SetIdentifierKey = normalizePersistedDNSRecordOptionalValue(func() string {
		if record.SetIdentifier == nil {
			return ""
		}
		return *record.SetIdentifier
	}())
	record.RegionKey = normalizePersistedDNSRecordStringKey(record.Region, false)
	record.FailoverKey = normalizePersistedDNSRecordStringKey(record.Failover, true)

	return dnsRecordPersistIdentity{
		DomainAssetID:    record.DomainAssetID,
		NameKey:          record.NameKey,
		TypeKey:          record.TypeKey,
		SetIdentifierKey: record.SetIdentifierKey,
		RegionKey:        record.RegionKey,
		FailoverKey:      record.FailoverKey,
	}, nil
}

// ExecuteSync runs the sync for an existing sync_job (used by queue workers)
// This method updates the sync_job status as it progresses
func (s *CloudAccountService) ExecuteSync(ctx context.Context, syncJobID uint64, cloudAccountID uint64) error {
	if s.accounts == nil || s.jobs == nil {
		return errors.New("sync pipeline is not configured")
	}

	// Get the sync job
	job, err := s.jobs.FindByID(ctx, syncJobID)
	if err != nil {
		return fmt.Errorf("failed to find sync job: %w", err)
	}

	// Verify it's for the correct account
	if job.CloudAccountID == nil || *job.CloudAccountID != cloudAccountID {
		return fmt.Errorf("sync job %d does not match cloud account %d", syncJobID, cloudAccountID)
	}
	if err := s.blockIfKillSwitchEnabled(ctx, job, cloudAccountID, "worker_dispatch"); err != nil {
		var killSwitchErr *KillSwitchError
		if errors.As(err, &killSwitchErr) {
			return nil
		}
		return err
	}

	// Get the account
	account, err := s.accounts.FindByID(ctx, cloudAccountID)
	if err != nil {
		return fmt.Errorf("failed to find cloud account: %w", err)
	}

	// Run the sync with a managed execution context (runSync updates status)
	s.runSyncWithExecutionContext(ctx, job, account, "queue_worker")

	return s.queueRetryErrorForExecutedJob(ctx, job)
}

// ExecuteSyncPayload executes a queue payload, tolerating legacy messages without SyncJobID.
// When SyncJobID is nil, it will create a sync_job record and run inline to avoid enqueue loops.
func (s *CloudAccountService) ExecuteSyncPayload(ctx context.Context, payload *queue.SyncJobPayload) error {
	if s.accounts == nil || s.jobs == nil {
		return errors.New("sync pipeline is not configured")
	}
	if payload == nil {
		return errors.New("empty sync payload")
	}
	if payload.SyncJobID == nil {
		if err := s.blockIfKillSwitchEnabled(ctx, nil, payload.CloudAccountID, "legacy_payload_dispatch"); err != nil {
			var killSwitchErr *KillSwitchError
			if errors.As(err, &killSwitchErr) {
				return nil
			}
			return err
		}
	}

	// Preferred path: sync_job already exists
	if payload.SyncJobID != nil {
		return s.ExecuteSync(ctx, *payload.SyncJobID, payload.CloudAccountID)
	}

	// Legacy payload without SyncJobID: create a job record and run inline (no re-enqueue to avoid recursion)
	account, job, err := s.createSyncJobRecord(ctx, payload.CloudAccountID, payload.TriggeredBy)
	if err != nil {
		return fmt.Errorf("failed to create sync job for legacy payload: %w", err)
	}
	runCtx := context.Background()
	s.runSyncWithExecutionContext(runCtx, job, account, "legacy_payload_inline")
	return s.queueRetryErrorForExecutedJob(runCtx, job)
}

// UpdateSyncJobStatus is a helper to update sync_job status (used by workers)
func (s *CloudAccountService) UpdateSyncJobStatus(ctx context.Context, syncJobID uint64, status caasmmodel.SyncJobStatus) error {
	if s.jobs == nil {
		return errors.New("sync job repository not configured")
	}

	job, err := s.jobs.FindByID(ctx, syncJobID)
	if err != nil {
		return err
	}

	job.Status = status
	return s.jobs.Update(ctx, job)
}

// SyncJobRepository interface (ensure it has FindByID)
// This is just for reference - the actual interface should be in repository package
type syncJobRepositoryInterface interface {
	Create(ctx context.Context, job *caasmmodel.SyncJob) error
	Update(ctx context.Context, job *caasmmodel.SyncJob) error
	FindByID(ctx context.Context, id uint64) (*caasmmodel.SyncJob, error)
	List(ctx context.Context, filter repository.SyncJobFilter, page, pageSize int) ([]caasmmodel.SyncJob, int64, error)
}

func (s *CloudAccountService) flattenDNSRecords(domainAsset *caasmmodel.DomainAsset) []caasmmodel.DNSRecord {
	if domainAsset == nil || domainAsset.DNSRecords == nil || domainAsset.ID == 0 {
		return nil
	}

	rawRecords, ok := domainAsset.DNSRecords["records"]
	if !ok {
		return nil
	}

	toInt := func(v interface{}) (int, bool) {
		switch val := v.(type) {
		case int:
			return val, true
		case int64:
			return int(val), true
		case float64:
			return int(val), true
		case float32:
			return int(val), true
		case string:
			if val == "" {
				return 0, false
			}
			if n, err := strconv.Atoi(val); err == nil {
				return n, true
			}
		}
		return 0, false
	}

	normalizeValues := func(v interface{}) []string {
		switch arr := v.(type) {
		case []string:
			return arr
		case []interface{}:
			out := make([]string, 0, len(arr))
			for _, item := range arr {
				switch sv := item.(type) {
				case string:
					out = append(out, sv)
				default:
					out = append(out, fmt.Sprint(sv))
				}
			}
			return out
		default:
			return nil
		}
	}

	var records []caasmmodel.DNSRecord

	switch recs := rawRecords.(type) {
	case []map[string]interface{}:
		for _, rec := range recs {
			if r := s.convertDNSRecord(domainAsset, rec, normalizeValues, toInt); r != nil {
				records = append(records, *r)
			}
		}
	case []interface{}:
		for _, raw := range recs {
			if rec, ok := raw.(map[string]interface{}); ok {
				if r := s.convertDNSRecord(domainAsset, rec, normalizeValues, toInt); r != nil {
					records = append(records, *r)
				}
			}
		}
	}
	return records
}

func (s *CloudAccountService) dnsRecordsForReplace(domainAsset *caasmmodel.DomainAsset) ([]caasmmodel.DNSRecord, bool) {
	if domainAsset == nil || domainAsset.DNSRecords == nil || domainAsset.ID == 0 {
		return nil, false
	}
	rawRecords, ok := domainAsset.DNSRecords["records"]
	if !ok {
		return nil, false
	}
	switch rawRecords.(type) {
	case []map[string]interface{}, []interface{}:
		return s.flattenDNSRecords(domainAsset), true
	default:
		return nil, false
	}
}

func (s *CloudAccountService) queueRetryErrorForExecutedJob(ctx context.Context, job *caasmmodel.SyncJob) error {
	if s.jobs == nil || job == nil || job.ID == 0 {
		return nil
	}
	latest, err := s.jobs.FindByID(syncJobDurableContext(ctx), job.ID)
	if err != nil {
		return fmt.Errorf("reload sync job %d after execution: %w", job.ID, err)
	}
	if latest == nil || latest.Status != caasmmodel.SyncJobFailed {
		return nil
	}
	if !isQueueRetryableFailedSyncJob(latest) {
		return nil
	}
	reason, _ := latest.Metadata["termination_reason"].(string)
	return &retryableSyncJobError{
		SyncJobID: latest.ID,
		Status:    latest.Status,
		Reason:    reason,
		Message:   latest.ErrorMessage,
	}
}

func isQueueRetryableFailedSyncJob(job *caasmmodel.SyncJob) bool {
	if job == nil || job.Status != caasmmodel.SyncJobFailed {
		return false
	}
	if strings.Contains(strings.ToLower(job.ErrorMessage), strings.ToLower(defaultKillSwitchMessage)) {
		return false
	}
	metadata := job.Metadata
	terminationReason, _ := metadata["termination_reason"].(string)
	cancellationSource, _ := metadata["cancellation_source"].(string)
	coordinationState, _ := metadata["coordination_state"].(string)
	switch strings.TrimSpace(strings.ToLower(terminationReason)) {
	case "partial_failures", "job_stopped", "external_cancellation":
		return false
	case "execution_timeout", "persist_error", "sync_error", "checkpoint_metadata_error":
		return true
	}
	switch strings.TrimSpace(strings.ToLower(cancellationSource)) {
	case "user_stop", "kill_switch":
		return false
	}
	switch strings.TrimSpace(strings.ToLower(coordinationState)) {
	case "account_overlap", "db_active_account_guard", "runtime_capacity_guard":
		return false
	}
	return strings.TrimSpace(job.ErrorMessage) != ""
}

func (s *CloudAccountService) resolveCollectionScope(ctx context.Context) resolvedCollectionScope {
	scope := resolvedCollectionScope{
		IPScope:     caasmmodel.CollectionScopeAll,
		PortScope:   caasmmodel.CollectionScopeAll,
		DomainScope: caasmmodel.CollectionScopeAll,
	}

	if s.config == nil {
		return scope
	}

	cfg, err := s.config.Get(ctx)
	if err != nil || cfg.Data == nil {
		return scope
	}

	var configData caasmmodel.ConfigData
	jsonBytes, err := json.Marshal(cfg.Data)
	if err != nil {
		return scope
	}
	if err := json.Unmarshal(jsonBytes, &configData); err != nil || configData.CollectionScope == nil {
		return scope
	}

	if configData.CollectionScope.IPCollectionScope != "" {
		scope.IPScope = configData.CollectionScope.IPCollectionScope
	}
	if configData.CollectionScope.PortCollectionScope != "" {
		scope.PortScope = configData.CollectionScope.PortCollectionScope
	}
	if configData.CollectionScope.DomainCollectionScope != "" {
		scope.DomainScope = configData.CollectionScope.DomainCollectionScope
	}

	return scope
}

func (s *CloudAccountService) filterSyncPayloadByCollectionScope(ctx context.Context, payload *caasmmodel.IngestPayload) {
	if payload == nil {
		return
	}

	scope := s.resolveCollectionScope(ctx)

	if scope.IPScope == caasmmodel.CollectionScopePublicOnly {
		filtered := make([]caasmmodel.IPAsset, 0, len(payload.IPAssets))
		for _, ipAsset := range payload.IPAssets {
			if ipAsset.IsPublic {
				filtered = append(filtered, ipAsset)
			}
		}
		payload.IPAssets = filtered
	}

	if scope.PortScope == caasmmodel.CollectionScopePublicOnly {
		filtered := make([]caasmmodel.IPPortIndex, 0, len(payload.Ports))
		for _, port := range payload.Ports {
			if strings.EqualFold(port.Exposure, "public") {
				filtered = append(filtered, port)
			}
		}
		payload.Ports = filtered
	}

	if scope.DomainScope == caasmmodel.CollectionScopePublicOnly {
		filtered := make([]caasmmodel.DomainAsset, 0, len(payload.DomainAssets))
		for _, domain := range payload.DomainAssets {
			if isDomainVisibleInPublicOnlyScope(domain) {
				filtered = append(filtered, domain)
			}
		}
		payload.DomainAssets = filtered
	}
}

func (s *CloudAccountService) prunePersistedDataByCollectionScope(ctx context.Context, cloudAccountID uint64) error {
	scope := s.resolveCollectionScope(ctx)

	if scope.IPScope == caasmmodel.CollectionScopePublicOnly && s.assets != nil {
		if err := s.assets.DeletePrivateIPAssetsForCloudAccount(ctx, cloudAccountID); err != nil {
			return fmt.Errorf("delete private ip assets for cloud account %d: %w", cloudAccountID, err)
		}
	}

	if scope.PortScope == caasmmodel.CollectionScopePublicOnly && s.ports != nil {
		if err := s.ports.DeleteNonPublicForCloudAccount(ctx, cloudAccountID); err != nil {
			return fmt.Errorf("delete non-public ports for cloud account %d: %w", cloudAccountID, err)
		}
	}

	if scope.DomainScope == caasmmodel.CollectionScopePublicOnly && s.assets != nil {
		if err := s.assets.DeletePrivateDomainAssetsForCloudAccount(ctx, cloudAccountID); err != nil {
			return fmt.Errorf("delete private domain assets for cloud account %d: %w", cloudAccountID, err)
		}
	}

	return nil
}

func isDomainVisibleInPublicOnlyScope(domain caasmmodel.DomainAsset) bool {
	zoneType := strings.ToLower(strings.TrimSpace(domain.ZoneType))
	return zoneType == "public" || zoneType == "public-and-private"
}

func (s *CloudAccountService) convertDNSRecord(
	domainAsset *caasmmodel.DomainAsset,
	rec map[string]interface{},
	normalizeValues func(v interface{}) []string,
	toInt func(v interface{}) (int, bool),
) *caasmmodel.DNSRecord {
	if rec == nil || domainAsset == nil {
		return nil
	}

	name := strings.TrimSuffix(fmt.Sprint(rec["name"]), ".")
	recordType := strings.ToUpper(fmt.Sprint(rec["type"]))

	ttl, _ := toInt(rec["ttl"])
	values := normalizeValues(rec["values"])

	var weightPtr *int
	if w, ok := toInt(rec["weight"]); ok {
		weightPtr = &w
	}

	var setIDPtr *string
	if sid, ok := rec["set_identifier"].(string); ok && sid != "" {
		setIDPtr = &sid
	}

	region := fmt.Sprint(rec["region"])
	if region == "<nil>" {
		region = ""
	}
	failover := fmt.Sprint(rec["failover"])
	if failover == "<nil>" {
		failover = ""
	}

	var alias coremodel.JSONB
	if aliasMap, ok := rec["alias_target"].(map[string]interface{}); ok && len(aliasMap) > 0 {
		alias = coremodel.JSONB(aliasMap)
	}

	var evalTarget *bool
	if alias != nil {
		if val, ok := alias["evaluate_target_health"].(bool); ok {
			evalTarget = &val
		}
	}

	var healthCheck string
	if hc, ok := rec["health_check_id"].(string); ok {
		healthCheck = hc
	}

	status := ""
	if val, ok := rec["status"].(string); ok && val != "" {
		status = strings.ToLower(val)
	}
	if enabled, ok := rec["enabled"].(bool); ok {
		if !enabled {
			status = "inactive"
		} else if status == "" {
			status = "active"
		}
	}
	if active, ok := rec["active"].(bool); ok {
		if !active {
			status = "inactive"
		} else if status == "" {
			status = "active"
		}
	}
	if status == "" && domainAsset.Status != "" {
		status = domainAsset.Status
	}
	if status == "" {
		status = "active"
	}

	isPublic := strings.ToLower(domainAsset.ZoneType) != "private"

	return &caasmmodel.DNSRecord{
		DomainAssetID: domainAsset.ID,
		Name:          name,
		Type:          recordType,
		TTL:           ttl,
		Values:        coremodel.StringArray(values),
		Weight:        weightPtr,
		SetIdentifier: setIDPtr,
		Region:        region,
		Failover:      failover,
		HealthCheckID: healthCheck,
		EvalTarget:    evalTarget,
		AliasTarget:   alias,
		Status:        status,
		IsPublic:      isPublic,
	}
}

func normalizeDNSValues(recordType string, values coremodel.StringArray, alias coremodel.JSONB) coremodel.StringArray {
	var out []string
	for _, v := range values {
		val := strings.TrimSpace(v)
		val = strings.TrimSuffix(val, ".")
		switch strings.ToUpper(recordType) {
		case "CNAME", "TXT", "MX", "NS", "SRV":
			val = strings.ToLower(val)
		}
		if val != "" {
			out = append(out, val)
		}
	}
	// Alias target fallback
	if len(out) == 0 && alias != nil {
		if dn, ok := alias["dns_name"].(string); ok && dn != "" {
			val := strings.TrimSuffix(strings.ToLower(strings.TrimSpace(dn)), ".")
			out = append(out, val)
		}
	}
	sort.Strings(out)
	return coremodel.StringArray(out)
}

func stringArrayEqual(a, b coremodel.StringArray) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

func shouldSuppressWithPattern(pattern *string, values coremodel.StringArray) bool {
	if pattern == nil || *pattern == "" {
		return false
	}
	re, err := regexp.Compile(*pattern)
	if err != nil {
		return false
	}
	for _, v := range values {
		if re.MatchString(v) {
			return true
		}
	}
	return false
}

func shouldRateLimitChange(lastAt *time.Time, lastTo coremodel.StringArray, newVals coremodel.StringArray) bool {
	if lastAt == nil {
		return false
	}
	// If we already alerted on the same target within the last hour, suppress.
	if time.Since(*lastAt) < time.Hour && stringArrayEqual(lastTo, newVals) {
		return true
	}
	return false
}

const persistNativeIDPrefetchChunkSize = 500
const persistDBRetryAttempts = 3

var persistDBRetryBackoffs = []time.Duration{100 * time.Millisecond, 250 * time.Millisecond}

func (s *CloudAccountService) persistSyncPayload(ctx context.Context, job *caasmmodel.SyncJob, payload *caasmmodel.IngestPayload) (err error) {
	if payload == nil {
		return errors.New("empty sync payload")
	}

	cloudAccountID := payload.CloudAccountID
	jobID := uint64(0)
	if job != nil {
		jobID = job.ID
	}
	nativeToAssetID := make(map[string]uint64)
	prefetchNativeIDs := collectPersistNativeIDs(payload)
	persistMetadata := coremodel.JSONB{
		"persist_asset_count":              len(payload.Assets),
		"persist_prefetch_native_id_count": len(prefetchNativeIDs),
		"persist_prefetch_chunk_count":     persistPrefetchChunkCount(len(prefetchNativeIDs)),
		"persist_prefetch_hit_count":       0,
		"persist_prefetch_miss_count":      len(prefetchNativeIDs),
	}
	defer func() {
		if err == nil {
			return
		}
		failureMetadata := coremodel.JSONB{
			"failed_phase":                     "persist",
			"persist_asset_count":              persistMetadata["persist_asset_count"],
			"persist_prefetch_native_id_count": persistMetadata["persist_prefetch_native_id_count"],
			"persist_prefetch_chunk_count":     persistMetadata["persist_prefetch_chunk_count"],
			"persist_prefetch_hit_count":       persistMetadata["persist_prefetch_hit_count"],
			"persist_prefetch_miss_count":      persistMetadata["persist_prefetch_miss_count"],
		}
		if isPersistContextError(err) && ctx.Err() == nil {
			failureMetadata["cancellation_source"] = derivePersistCancellationSource(err)
		}
		s.updateSyncJobMetadata(ctx, job, failureMetadata)
		if job != nil {
			s.recordJobEvent(ctx, job.ID, "error", "Persist failed with write errors", map[string]interface{}{
				"phase":         "persist",
				"failure_count": 1,
				"error":         fmt.Sprintf("persist: %v", err),
			})
		}
	}()
	if len(prefetchNativeIDs) > 0 {
		var existingAssets []repository.AssetMinimal
		lookupErr := retryPersistDBOperation(ctx, func(opCtx context.Context) error {
			var err error
			existingAssets, err = s.assets.FindMinimalByNativeIDs(opCtx, cloudAccountID, prefetchNativeIDs)
			return err
		})
		if lookupErr != nil {
			return fmt.Errorf("prefetch existing assets for cloud account %d: %w", cloudAccountID, lookupErr)
		}
		persistMetadata["persist_prefetch_hit_count"] = len(existingAssets)
		persistMetadata["persist_prefetch_miss_count"] = len(prefetchNativeIDs) - len(existingAssets)
		for _, existing := range existingAssets {
			nativeID := strings.TrimSpace(existing.NativeID)
			if nativeID == "" {
				continue
			}
			if existing.ID != 0 {
				nativeToAssetID[nativeID] = existing.ID
			}
		}
	}
	s.updateSyncJobMetadata(ctx, job, persistMetadata)
	prepareAssetForSave := func(asset *caasmmodel.Asset) error {
		if asset == nil {
			return errors.New("asset is nil")
		}
		if asset.CloudAccountID == nil || *asset.CloudAccountID == 0 {
			asset.CloudAccountID = &cloudAccountID
		}
		if asset.CloudAccountID == nil || *asset.CloudAccountID == 0 {
			return repository.ErrAssetCloudAccountIDRequired
		}
		asset.NativeID = strings.TrimSpace(asset.NativeID)
		if asset.NativeID == "" {
			return repository.ErrAssetNativeIDRequired
		}
		asset.OwnerEmployeeID = nil
		asset.DepartmentID = nil
		return nil
	}
	stripManualOwnerFields := func(asset *caasmmodel.Asset) {
		if asset == nil {
			return
		}
		asset.OwnerEmployeeID = nil
		asset.DepartmentID = nil
	}
	assetHasMaterialFields := func(asset *caasmmodel.Asset) bool {
		if asset == nil {
			return false
		}
		return strings.TrimSpace(asset.Name) != "" ||
			strings.TrimSpace(string(asset.AssetType)) != "" ||
			strings.TrimSpace(asset.Service) != "" ||
			strings.TrimSpace(asset.Criticality) != "" ||
			strings.TrimSpace(string(asset.Status)) != "" ||
			strings.TrimSpace(asset.Environment) != "" ||
			strings.TrimSpace(asset.Region) != "" ||
			strings.TrimSpace(asset.Provider) != "" ||
			len(asset.Metadata) > 0
	}
	assetHasSnapshotIdentityFields := func(asset *caasmmodel.Asset) bool {
		if asset == nil {
			return false
		}
		return strings.TrimSpace(asset.Name) != "" ||
			strings.TrimSpace(string(asset.AssetType)) != ""
	}
	resolveAssetID := func(asset *caasmmodel.Asset) (uint64, error) {
		if err := prepareAssetForSave(asset); err != nil {
			return 0, err
		}
		if !assetHasSnapshotIdentityFields(asset) {
			if id, ok := nativeToAssetID[asset.NativeID]; ok {
				asset.ID = id
				stripManualOwnerFields(asset)
				return id, nil
			}
		}
		if assetHasMaterialFields(asset) {
			if err := retryPersistDBOperation(ctx, func(opCtx context.Context) error {
				return s.assets.SaveAssetSnapshot(opCtx, asset)
			}); err != nil {
				if isPersistContextError(err) {
					return 0, err
				}
				if errors.Is(err, repository.ErrAssetCloudAccountIDRequired) || errors.Is(err, repository.ErrAssetNativeIDRequired) {
					return 0, err
				}
				return 0, fmt.Errorf("save asset %s: %w", asset.NativeID, err)
			}
			if asset.ID == 0 {
				return 0, fmt.Errorf("save asset %s returned empty id", asset.NativeID)
			}
			updatePersistAssetIDLookupMap(nativeToAssetID, asset)
			return asset.ID, nil
		}
		if id, ok := nativeToAssetID[asset.NativeID]; ok {
			asset.ID = id
			stripManualOwnerFields(asset)
			return id, nil
		}
		return 0, fmt.Errorf("resolve reference-only asset %s for cloud account %d returned empty asset", asset.NativeID, cloudAccountID)
	}

	// 1. Save Assets
	for _, asset := range payload.Assets {
		if err := prepareAssetForSave(&asset); err != nil {
			return err
		}
		if !assetHasMaterialFields(&asset) {
			if id, ok := nativeToAssetID[asset.NativeID]; ok {
				asset.ID = id
			} else {
				if id, ok := nativeToAssetID[asset.NativeID]; !ok || id == 0 {
					return fmt.Errorf("resolve reference-only asset %s for cloud account %d returned empty asset", asset.NativeID, cloudAccountID)
				}
				asset.ID = nativeToAssetID[asset.NativeID]
			}
			stripManualOwnerFields(&asset)
			continue
		}
		if err := retryPersistDBOperation(ctx, func(opCtx context.Context) error {
			return s.assets.SaveAssetSnapshot(opCtx, &asset)
		}); err != nil {
			if isPersistContextError(err) {
				return err
			}
			if errors.Is(err, repository.ErrAssetCloudAccountIDRequired) || errors.Is(err, repository.ErrAssetNativeIDRequired) {
				return err
			}
			return fmt.Errorf("asset.save %s: %w", asset.NativeID, err)
		}
		updatePersistAssetIDLookupMap(nativeToAssetID, &asset)
		if err := s.recordHistory(ctx, asset.ID, "upsert", nil); err != nil {
			return fmt.Errorf("history.asset_upsert %d: %w", asset.ID, err)
		}

		if s.tags != nil && asset.Metadata != nil {
			if tagMap, ok := extractTags(asset.Metadata); ok {
				if err := s.tags.ReplaceTags(ctx, asset.ID, tagMap); err != nil {
					return fmt.Errorf("asset.tags.replace %d: %w", asset.ID, err)
				} else {
					if err := s.recordHistory(ctx, asset.ID, "tags.update", coremodel.JSONB{"after": tagMap}); err != nil {
						return fmt.Errorf("history.asset_tags_update %d: %w", asset.ID, err)
					}
				}
			}
		}
	}

	// 2. Save IP Assets
	ipAssetMap := make(map[string][]uint64)
	for _, ipAsset := range payload.IPAssets {
		if ipAsset.AssetID == 0 && ipAsset.Asset.NativeID != "" {
			assetID, err := resolveAssetID(&ipAsset.Asset)
			if err != nil {
				return err
			}
			ipAsset.AssetID = assetID
		}

		if ipAsset.AssetID == 0 {
			s.logWarn("skipping IP asset due to missing asset id",
				"job_id", jobID,
				"cloud_account_id", cloudAccountID,
				"phase", "persist_ip_assets",
				"ip_address", ipAsset.IPAddress,
			)
			continue
		}
		// Persist the owning asset by FK only. Leaving the transient Asset payload attached
		// can cause GORM to synthesize an empty assets row during Save.
		ipAsset.Asset = caasmmodel.Asset{}

		if err := retryPersistDBOperation(ctx, func(opCtx context.Context) error {
			return s.assets.SaveIPAsset(opCtx, &ipAsset)
		}); err != nil {
			if isPersistContextError(err) {
				return err
			}
			return fmt.Errorf("ip_asset.save %s: %w", ipAsset.IPAddress, err)
		}
		if ipAsset.AssetID != 0 {
			ipAssetMap[ipAsset.IPAddress] = appendUniqueUint64(ipAssetMap[ipAsset.IPAddress], ipAsset.AssetID)
		}
		if err := s.recordHistory(ctx, ipAsset.AssetID, "ip_upsert", nil); err != nil {
			return fmt.Errorf("history.ip_upsert %d: %w", ipAsset.AssetID, err)
		}
	}

	// 3. Save Domain Assets
	domainAssetMap := make(map[string][]uint64)
	for _, domainAsset := range payload.DomainAssets {
		if domainAsset.AssetID == 0 && domainAsset.Asset.NativeID != "" {
			assetID, err := resolveAssetID(&domainAsset.Asset)
			if err != nil {
				return err
			}
			domainAsset.AssetID = assetID
		}

		if domainAsset.Asset.Service == "" {
			if strings.Contains(strings.ToLower(domainAsset.Registrar), "route53") {
				domainAsset.Asset.Service = "Route53"
			} else if strings.Contains(strings.ToLower(domainAsset.Registrar), "s3") {
				domainAsset.Asset.Service = "S3"
			}
			if domainAsset.Asset.Service != "" {
				s.logInfo("auto-populated domain asset service",
					"job_id", jobID,
					"cloud_account_id", cloudAccountID,
					"phase", "persist_domain_assets",
					"domain_name", domainAsset.DomainName,
					"service", domainAsset.Asset.Service,
				)
			}
		}

		if domainAsset.AssetID == 0 {
			s.logWarn("skipping domain asset due to missing asset id",
				"job_id", jobID,
				"cloud_account_id", cloudAccountID,
				"phase", "persist_domain_assets",
				"domain_name", domainAsset.DomainName,
			)
			continue
		}
		if domainAsset.DomainClass == "" {
			domainAsset.DomainClass = caasmmodel.DomainClassUnknown
		}
		stripManualOwnerFields(&domainAsset.Asset)
		domainAsset.OwnerEmployeeID = nil
		domainAsset.DepartmentID = nil
		// SaveDomainAsset uses raw SQL keyed by AssetID/domain, so drop the transient association
		// to avoid accidental follow-on ORM writes against an incomplete Asset payload.
		domainAsset.Asset = caasmmodel.Asset{}

		if err := retryPersistDBOperation(ctx, func(opCtx context.Context) error {
			return s.assets.SaveDomainAssetSnapshot(opCtx, &domainAsset)
		}); err != nil {
			if isPersistContextError(err) {
				return err
			}
			return fmt.Errorf("domain_asset.save %s: %w", domainAsset.DomainName, err)
		}

		if s.dnsRecords != nil && domainAsset.ID != 0 {
			records, shouldReplaceDNSRecords := s.dnsRecordsForReplace(&domainAsset)
			if !shouldReplaceDNSRecords {
				if domainAsset.DNSRecords == nil {
					s.logInfo("skipping DNS record replace; records were not collected",
						"job_id", jobID,
						"cloud_account_id", cloudAccountID,
						"phase", "persist_domain_assets",
						"domain_name", domainAsset.DomainName,
						"reason", "dns_records_missing",
					)
				}
			} else {
				var existing []caasmmodel.DNSRecord
				if err := retryPersistDBOperation(ctx, func(opCtx context.Context) error {
					var err error
					existing, err = s.dnsRecords.GetByDomain(opCtx, domainAsset.ID)
					return err
				}); err != nil {
					if isPersistContextError(err) {
						return fmt.Errorf("load existing dns records for domain %s: %w", domainAsset.DomainName, err)
					}
					return fmt.Errorf("dns_records.load_existing %s: %w", domainAsset.DomainName, err)
				}
				existingByKey := make(map[dnsRecordPersistIdentity]caasmmodel.DNSRecord)
				for _, er := range existing {
					key, err := normalizePersistedDNSRecordIdentity(&er)
					if err != nil {
						return fmt.Errorf("normalize existing dns record identity for domain %s: %w", domainAsset.DomainName, err)
					}
					existingByKey[key] = er
				}
				seenChanged := make(map[dnsRecordPersistIdentity]bool)
				for i := range records {
					rec := &records[i]
					key, err := normalizePersistedDNSRecordIdentity(rec)
					if err != nil {
						return fmt.Errorf("normalize dns record identity for domain %s: %w", domainAsset.DomainName, err)
					}
					if prev, ok := existingByKey[key]; ok {
						if rec.OwnerDepartmentID == "" {
							rec.OwnerDepartmentID = prev.OwnerDepartmentID
						}
						if rec.OwnerEmail == "" {
							rec.OwnerEmail = prev.OwnerEmail
						}
						if rec.OwnerEmployeeID == nil {
							rec.OwnerEmployeeID = prev.OwnerEmployeeID
						}
						rec.IgnoreChanges = prev.IgnoreChanges
						rec.SuppressPattern = prev.SuppressPattern
						rec.LastChangeAt = prev.LastChangeAt
						rec.LastChangeFrom = prev.LastChangeFrom
						rec.LastChangeTo = prev.LastChangeTo

						oldVals := normalizeDNSValues(prev.Type, prev.Values, prev.AliasTarget)
						newVals := normalizeDNSValues(rec.Type, rec.Values, rec.AliasTarget)
						if !stringArrayEqual(oldVals, newVals) {
							rec.LastChangeFrom = oldVals
							rec.LastChangeTo = newVals
							now := time.Now()
							rec.LastChangeAt = &now
							if rec.IgnoreChanges || shouldSuppressWithPattern(&rec.SuppressPattern, newVals) {
								if job != nil {
									s.recordJobEvent(ctx, job.ID, "info", "DNS target change suppressed", map[string]interface{}{
										"zone":             domainAsset.DomainName,
										"record":           rec.Name,
										"type":             rec.Type,
										"old_values":       []string(oldVals),
										"new_values":       []string(newVals),
										"ignore_changes":   rec.IgnoreChanges,
										"suppress_pattern": rec.SuppressPattern,
									})
								}
								continue
							}
							if shouldRateLimitChange(prev.LastChangeAt, prev.LastChangeTo, newVals) || seenChanged[key] {
								continue
							}
							seenChanged[key] = true
							if job != nil {
								s.recordJobEvent(ctx, job.ID, "warning", "DNS target changed", map[string]interface{}{
									"zone":       domainAsset.DomainName,
									"record":     rec.Name,
									"type":       rec.Type,
									"old_values": []string(oldVals),
									"new_values": []string(newVals),
								})
							}
						}
					}
				}
				if err := retryPersistDBOperation(ctx, func(opCtx context.Context) error {
					return s.dnsRecords.ReplaceForDomain(opCtx, domainAsset.ID, records)
				}); err != nil {
					if isPersistContextError(err) {
						return fmt.Errorf("sync dns records for domain %s: %w", domainAsset.DomainName, err)
					}
					return fmt.Errorf("dns_records.replace %s: %w", domainAsset.DomainName, err)
				}
			}
		}

		if domainAsset.AssetID != 0 {
			domainAssetMap[domainAsset.DomainName] = appendUniqueUint64(domainAssetMap[domainAsset.DomainName], domainAsset.AssetID)
		}
		if err := s.recordHistory(ctx, domainAsset.AssetID, "domain_upsert", nil); err != nil {
			return fmt.Errorf("history.domain_upsert %d: %w", domainAsset.AssetID, err)
		}
	}

	// 4. Save Port Index
	if s.ports != nil {
		for _, port := range payload.Ports {
			port.NormalizeEndpointFields()
			endpoint := port.EndpointValue()
			assetIDs := []uint64{}
			if port.AssetID != 0 {
				assetIDs = append(assetIDs, port.AssetID)
			} else if port.Asset.NativeID != "" {
				assetID, err := resolveAssetID(&port.Asset)
				if err != nil {
					return err
				}
				assetIDs = append(assetIDs, assetID)
			} else {
				assetIDs = append(assetIDs, ipAssetMap[port.IPAddress]...)
				assetIDs = append(assetIDs, domainAssetMap[endpoint]...)
				assetIDs = uniqueUint64s(assetIDs)
				if len(assetIDs) == 0 {
					s.logWarn("skipping port due to unknown asset id",
						"job_id", jobID,
						"cloud_account_id", cloudAccountID,
						"phase", "persist_ports",
						"endpoint", endpoint,
						"ip_address", port.IPAddress,
						"port", port.Port,
						"protocol", port.Protocol,
					)
					continue
				}
			}
			if port.LastSeenAt.IsZero() {
				port.LastSeenAt = time.Now()
			}
			if port.ObservationSource == "" {
				port.ObservationSource = caasmmodel.PortObservationSourceUnknown
			}
			for _, assetID := range assetIDs {
				portForAsset := port
				portForAsset.AssetID = assetID
				// Persist the resolved owning asset by FK only. Keeping the collector's transient
				// Asset payload attached lets GORM insert blank asset rows on Create.
				portForAsset.Asset = caasmmodel.Asset{}
				if err := retryPersistDBOperation(ctx, func(opCtx context.Context) error {
					return s.ports.Upsert(opCtx, &portForAsset)
				}); err != nil {
					if isPersistContextError(err) {
						return err
					}
					return fmt.Errorf("port.upsert %s:%d/%s for asset %d: %w", portForAsset.IPAddress, portForAsset.Port, portForAsset.Protocol, assetID, err)
				}
			}
		}
	}

	return nil
}

func updatePersistAssetIDLookupMap(nativeToAssetID map[string]uint64, asset *caasmmodel.Asset) {
	if asset == nil {
		return
	}
	nativeID := strings.TrimSpace(asset.NativeID)
	if nativeID == "" {
		return
	}
	if asset.ID != 0 {
		nativeToAssetID[nativeID] = asset.ID
	}
}

func collectPersistNativeIDs(payload *caasmmodel.IngestPayload) []string {
	if payload == nil {
		return nil
	}
	candidates := make([]string, 0, len(payload.Assets)+len(payload.IPAssets)+len(payload.DomainAssets)+len(payload.Ports))
	for _, asset := range payload.Assets {
		candidates = append(candidates, asset.NativeID)
	}
	for _, ipAsset := range payload.IPAssets {
		candidates = append(candidates, ipAsset.Asset.NativeID)
	}
	for _, domainAsset := range payload.DomainAssets {
		candidates = append(candidates, domainAsset.Asset.NativeID)
	}
	for _, port := range payload.Ports {
		candidates = append(candidates, port.Asset.NativeID)
	}
	return uniqueTrimmedStrings(candidates)
}

func uniqueTrimmedStrings(values []string) []string {
	if len(values) == 0 {
		return nil
	}
	seen := make(map[string]struct{}, len(values))
	unique := make([]string, 0, len(values))
	for _, value := range values {
		trimmed := strings.TrimSpace(value)
		if trimmed == "" {
			continue
		}
		if _, ok := seen[trimmed]; ok {
			continue
		}
		seen[trimmed] = struct{}{}
		unique = append(unique, trimmed)
	}
	sort.Strings(unique)
	return unique
}

func persistPrefetchChunkCount(nativeIDCount int) int {
	if nativeIDCount <= 0 {
		return 0
	}
	return (nativeIDCount + persistNativeIDPrefetchChunkSize - 1) / persistNativeIDPrefetchChunkSize
}

func derivePersistCancellationSource(err error) string {
	if errors.Is(err, context.DeadlineExceeded) {
		return "db_query_deadline_exceeded"
	}
	return "db_query_canceled"
}

// IngestSyncData processes data collected by an external serverless worker.
func (s *CloudAccountService) IngestSyncData(ctx context.Context, payload *caasmmodel.IngestPayload) error {
	if s.accounts == nil || s.jobs == nil || s.assets == nil {
		return errors.New("pipeline not fully configured")
	}

	// 1. Validate Sync Job
	job, err := s.jobs.FindByID(ctx, payload.SyncJobID)
	if err != nil {
		return fmt.Errorf("sync job not found: %w", err)
	}

	if isSyncJobTerminalStatus(job.Status) {
		s.recordJobEvent(syncJobDurableContext(ctx), job.ID, "warning", "Ingest skipped because sync job is already terminal", map[string]interface{}{
			"phase":             "ingest_start",
			"cloud_account_id":  payload.CloudAccountID,
			"current_status":    job.Status,
			"termination_guard": "terminal_status",
		})
		return nil
	}

	// Mark job running if not already.
	if job.Status != caasmmodel.SyncJobRunning {
		if !isSyncJobStartableStatus(job.Status) {
			return fmt.Errorf("invalid state transition: cannot ingest job in state %s", job.Status)
		}
		job.Status = caasmmodel.SyncJobRunning
		now := time.Now()
		job.StartedAt = &now
		if err := s.jobs.UpdateFields(ctx, job.ID, map[string]interface{}{
			"status":     job.Status,
			"started_at": now,
		}); err != nil {
			return err
		}
	}

	s.filterSyncPayloadByCollectionScope(ctx, payload)
	if err := s.prunePersistedDataByCollectionScope(ctx, payload.CloudAccountID); err != nil {
		return err
	}

	if err := s.persistSyncPayload(ctx, job, payload); err != nil {
		return err
	}

	// 5. Update Job Status
	job.Status = caasmmodel.SyncJobSuccess
	finishedAt := time.Now()
	job.FinishedAt = &finishedAt

	// Use UpdateFields to ensure status is persisted
	if err := s.jobs.UpdateFields(ctx, job.ID, map[string]interface{}{
		"status":      caasmmodel.SyncJobSuccess,
		"finished_at": finishedAt,
	}); err != nil {
		return err
	}

	// Clear backoff on success
	s.clearBackoff(payload.CloudAccountID)
	return nil
}

func appendUniqueUint64(values []uint64, next uint64) []uint64 {
	for _, value := range values {
		if value == next {
			return values
		}
	}
	return append(values, next)
}

func uniqueUint64s(values []uint64) []uint64 {
	if len(values) <= 1 {
		return values
	}

	unique := make([]uint64, 0, len(values))
	seen := make(map[uint64]struct{}, len(values))
	for _, value := range values {
		if _, ok := seen[value]; ok {
			continue
		}
		seen[value] = struct{}{}
		unique = append(unique, value)
	}
	return unique
}

func isPersistContextError(err error) bool {
	return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)
}

func retryPersistDBOperation(ctx context.Context, op func(context.Context) error) error {
	if op == nil {
		return errors.New("persist db operation is nil")
	}

	for attempt := 1; attempt <= persistDBRetryAttempts; attempt++ {
		if ctx != nil && ctx.Err() != nil {
			return ctx.Err()
		}

		err := op(ctx)
		if err == nil {
			return nil
		}
		if !isPersistContextError(err) || ctx == nil || ctx.Err() != nil || attempt == persistDBRetryAttempts {
			if ctx != nil && ctx.Err() != nil {
				return ctx.Err()
			}
			return err
		}

		backoff := persistDBRetryBackoffs[attempt-1]
		timer := time.NewTimer(backoff)
		select {
		case <-ctx.Done():
			if !timer.Stop() {
				<-timer.C
			}
			return ctx.Err()
		case <-timer.C:
		}
	}

	return nil
}