package service

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"

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

type stubCollector struct {
	result *collector.AssetResult
	err    error
}

type blockingCollector struct {
	started chan struct{}
	once    sync.Once
}

type interruptedChunkCollector struct {
	first     *collector.AssetChunk
	second    *collector.AssetChunk
	secondCtx context.Context
}

type stubConfigRepoInternal struct {
	cfg                *caasmmodel.Config
	err                error
	killSwitchSequence []bool
	callCount          int
}

type stubSyncJobEventRepo struct {
	events []*caasmmodel.SyncJobEvent
}

type stubDNSRecordRepoInternal struct {
	existing         []caasmmodel.DNSRecord
	replacedDomainID uint64
	replacedRecords  []caasmmodel.DNSRecord
}

func (s *stubConfigRepoInternal) Get(ctx context.Context) (*caasmmodel.Config, error) {
	if len(s.killSwitchSequence) > 0 {
		enabled := s.killSwitchSequence[s.callCount]
		if s.callCount < len(s.killSwitchSequence)-1 {
			s.callCount++
		}
		cfg := &caasmmodel.Config{}
		if s.cfg != nil {
			*cfg = *s.cfg
			if s.cfg.Data != nil {
				cfg.Data = coremodel.JSONB{}
				for key, value := range s.cfg.Data {
					cfg.Data[key] = value
				}
			}
		}
		if cfg.Data == nil {
			cfg.Data = coremodel.JSONB{}
		}
		cfg.Data["sync_settings"] = map[string]interface{}{
			"mode":                "queue",
			"kill_switch_enabled": enabled,
		}
		return cfg, s.err
	}
	return s.cfg, s.err
}

func (s *stubConfigRepoInternal) Upsert(ctx context.Context, data caasmmodel.Config) error {
	s.cfg = &data
	return s.err
}

func (r *stubSyncJobEventRepo) Create(ctx context.Context, event *caasmmodel.SyncJobEvent) error {
	r.events = append(r.events, event)
	return nil
}

func (r *stubSyncJobEventRepo) List(ctx context.Context, syncJobID uint64, opts repository.SyncJobEventListOptions) ([]caasmmodel.SyncJobEvent, error) {
	return nil, nil
}

func (r *stubDNSRecordRepoInternal) List(ctx context.Context, filter repository.DNSRecordFilter, page, pageSize int) ([]caasmmodel.DNSRecord, int64, error) {
	return nil, 0, nil
}

func (r *stubDNSRecordRepoInternal) GetByDomain(ctx context.Context, domainID uint64) ([]caasmmodel.DNSRecord, error) {
	records := make([]caasmmodel.DNSRecord, len(r.existing))
	copy(records, r.existing)
	return records, nil
}

func (r *stubDNSRecordRepoInternal) ReplaceForDomain(ctx context.Context, domainID uint64, records []caasmmodel.DNSRecord) error {
	r.replacedDomainID = domainID
	r.replacedRecords = append([]caasmmodel.DNSRecord(nil), records...)
	return nil
}

func (r *stubDNSRecordRepoInternal) UpdateOwner(ctx context.Context, id uint64, ownerEmail, deptID string, employeeID *uint64, ignoreChanges *bool, suppressPattern *string) error {
	return nil
}

func (c *stubCollector) Collect(ctx context.Context) (*collector.AssetResult, error) {
	if c.err != nil {
		return nil, c.err
	}
	if c.result == nil {
		return &collector.AssetResult{}, nil
	}
	return c.result, nil
}

func (c *stubCollector) Provider() string {
	return "aws"
}

func (c *stubCollector) SetEventReporter(reporter collector.EventReporter) {}

func (c *blockingCollector) Collect(ctx context.Context) (*collector.AssetResult, error) {
	c.once.Do(func() {
		if c.started != nil {
			close(c.started)
		}
	})
	<-ctx.Done()
	return nil, ctx.Err()
}

func (c *blockingCollector) Provider() string {
	return "aws"
}

func (c *blockingCollector) SetEventReporter(reporter collector.EventReporter) {}

func (c *interruptedChunkCollector) Collect(ctx context.Context) (*collector.AssetResult, error) {
	return nil, errors.New("unexpected Collect call in interrupted chunk test")
}

func (c *interruptedChunkCollector) Provider() string {
	return "aws"
}

func (c *interruptedChunkCollector) SetEventReporter(reporter collector.EventReporter) {}

func (c *interruptedChunkCollector) CollectChunks(ctx context.Context, sink collector.ChunkSink) (*collector.CollectionSummary, error) {
	summary := &collector.CollectionSummary{}
	if c.first != nil {
		if err := sink.PersistChunk(ctx, c.first); err != nil {
			summary.PartialFailures = append(summary.PartialFailures, collector.CollectionIssue{
				Scope:   c.first.Scope,
				Target:  c.first.Target,
				Message: err.Error(),
			})
		} else {
			summary.ChunkCount++
		}
	}
	if c.second != nil {
		persistCtx := ctx
		if c.secondCtx != nil {
			persistCtx = c.secondCtx
		}
		if err := sink.PersistChunk(persistCtx, c.second); err != nil {
			summary.PartialFailures = append(summary.PartialFailures, collector.CollectionIssue{
				Scope:   c.second.Scope,
				Target:  c.second.Target,
				Message: err.Error(),
			})
		} else {
			summary.ChunkCount++
		}
	}
	return summary, nil
}

func allowResumeCursorLookup(mockJobRepo *mocks.MockSyncJobRepository, ctx context.Context, accountID uint64) {
	mockJobRepo.On("FindLatestResumableJobForAccount", ctx, accountID, mock.AnythingOfType("uint64"), mock.AnythingOfType("string")).Return((*caasmmodel.SyncJob)(nil), nil).Maybe()
	mockJobRepo.On("FindByID", mock.Anything, mock.AnythingOfType("uint64")).Return(&caasmmodel.SyncJob{}, nil).Maybe()
}

func TestRunSyncSuccessfulRunMarksStaleAfterPersist(t *testing.T) {
	ctx := context.Background()
	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockJobRepo := new(mocks.MockSyncJobRepository)
	mockAssetRepo := new(mocks.MockAssetRepository)

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return &stubCollector{result: &collector.AssetResult{}}, "eu-west-2", nil
	}

	job := &caasmmodel.SyncJob{
		Base:     coremodel.Base{ID: 41},
		Metadata: coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 7},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}

	mockJobRepo.On("UpdateFields", ctx, uint64(41), mock.Anything).Return(nil)
	allowResumeCursorLookup(mockJobRepo, ctx, 7)
	mockAcctRepo.On("Update", ctx, mock.Anything).Return(nil).Once()
	mockAssetRepo.On("SoftDeleteStale", ctx, uint64(7), mock.Anything).Return(nil).Once()

	svc.runSync(ctx, job, account)

	assert.Equal(t, caasmmodel.SyncJobSuccess, job.Status)
	assert.Equal(t, "", job.ErrorMessage)
	assert.NotNil(t, job.FinishedAt)
	assert.NotNil(t, account.LastSyncAt)
	assert.Equal(t, "", account.ErrorMessage)
	assert.Equal(t, true, job.Metadata["collection_completed"])
	assert.Equal(t, true, job.Metadata["persist_completed"])
	assert.Equal(t, true, job.Metadata["stale_mark_completed"])
	assert.Equal(t, "", job.Metadata["stale_mark_error"])
	mockJobRepo.AssertExpectations(t)
	mockAcctRepo.AssertExpectations(t)
	mockAssetRepo.AssertExpectations(t)
}

func TestRunSyncCollectFailureBecomesPartialSuccessAndDoesNotMarkStale(t *testing.T) {
	ctx := context.Background()
	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockJobRepo := new(mocks.MockSyncJobRepository)
	mockAssetRepo := new(mocks.MockAssetRepository)
	eventRepo := &stubSyncJobEventRepo{}

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.events = eventRepo
	svc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return &stubCollector{err: errors.New("collect failed")}, "eu-west-2", nil
	}

	job := &caasmmodel.SyncJob{
		Base:     coremodel.Base{ID: 42},
		Metadata: coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 8},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}

	mockJobRepo.On("UpdateFields", ctx, uint64(42), mock.Anything).Return(nil)
	allowResumeCursorLookup(mockJobRepo, ctx, 8)
	mockAcctRepo.On("Update", ctx, mock.Anything).Return(nil).Once()

	svc.runSync(ctx, job, account)

	assert.Equal(t, caasmmodel.SyncJobPartial, job.Status)
	assert.Contains(t, job.ErrorMessage, "partial failures")
	assert.NotNil(t, job.FinishedAt)
	assert.Nil(t, account.LastSyncAt)
	assert.Contains(t, account.ErrorMessage, "partial failures")
	assert.Equal(t, false, job.Metadata["collection_completed"])
	assert.Equal(t, false, job.Metadata["persist_completed"])
	assert.Equal(t, false, job.Metadata["stale_mark_completed"])
	assert.Equal(t, "", job.Metadata["stale_mark_error"])
	assert.Equal(t, "collection", job.Metadata["failed_phase"])
	assert.Equal(t, 1, job.Metadata["partial_failure_count"])
	assert.Equal(t, true, job.Metadata["completed_with_warnings"])
	require.Len(t, eventRepo.events, 3)
	assert.Equal(t, "Collection degraded to partial failure", eventRepo.events[1].Message)
	assert.Equal(t, "collect failed", eventRepo.events[1].Metadata["error"])
	assert.Equal(t, "collection", eventRepo.events[1].Metadata["failed_phase"])
	assert.Equal(t, "sync completed with 1 partial failures", eventRepo.events[2].Message)
	assert.Equal(t, 1, eventRepo.events[2].Metadata["partial_failure_count"])
	mockAssetRepo.AssertNotCalled(t, "SoftDeleteStale", mock.Anything, mock.Anything, mock.Anything)
	mockJobRepo.AssertExpectations(t)
	mockAcctRepo.AssertExpectations(t)
}

func TestRunSyncExternalStopPreservesFailedStatusAndSkipsStaleMark(t *testing.T) {
	ctx := context.Background()
	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockJobRepo := new(mocks.MockSyncJobRepository)
	mockAssetRepo := new(mocks.MockAssetRepository)
	eventRepo := &stubSyncJobEventRepo{}

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.events = eventRepo
	blocker := &blockingCollector{started: make(chan struct{})}
	svc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return blocker, "eu-west-2", nil
	}

	job := &caasmmodel.SyncJob{
		Base:           coremodel.Base{ID: 46},
		CloudAccountID: func() *uint64 { v := uint64(12); return &v }(),
		Metadata:       coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 12},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}
	payload := &queue.SyncJobPayload{
		SyncJobID:      func() *uint64 { v := uint64(46); return &v }(),
		CloudAccountID: 12,
		Priority:       queue.PriorityNormal,
	}

	mockJobRepo.On("FindByID", mock.Anything, uint64(46)).Return(job, nil)
	mockJobRepo.On("UpdateFields", mock.Anything, uint64(46), mock.Anything).Return(nil)
	mockJobRepo.On("FindLatestResumableJobForAccount", mock.Anything, uint64(12), uint64(46), "").Return((*caasmmodel.SyncJob)(nil), nil).Maybe()
	mockAcctRepo.On("FindByID", mock.Anything, uint64(12)).Return(account, nil)
	mockAcctRepo.On("Update", mock.Anything, mock.Anything).Return(nil).Once()

	errCh := make(chan error, 1)
	go func() {
		errCh <- svc.invokeLambdaSimulated(ctx, "local-lightweight-worker", payload)
	}()

	select {
	case <-blocker.started:
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for blocking collector to start")
	}

	assert.True(t, svc.CancelSyncJob(46))

	select {
	case err := <-errCh:
		assert.NoError(t, err)
	case <-time.After(5 * time.Second):
		t.Fatal("timed out waiting for canceled sync to exit")
	}

	assert.Equal(t, caasmmodel.SyncJobFailed, job.Status)
	assert.Equal(t, "Job interrupted by external cancellation", job.ErrorMessage)
	assert.NotNil(t, job.FinishedAt)
	assert.Equal(t, false, job.Metadata["stale_mark_completed"])
	assert.Equal(t, false, job.Metadata["full_run_completed"])
	assert.Equal(t, false, job.Metadata["stale_mark_eligible"])
	assert.Equal(t, "sync_error", job.Metadata["stale_mark_skipped_reason"])
	assert.Equal(t, "external_cancellation", job.Metadata["termination_reason"])
	assert.Equal(t, "parent_context_canceled", job.Metadata["cancellation_source"])
	assert.Nil(t, account.LastSyncAt)
	mockAssetRepo.AssertNotCalled(t, "SoftDeleteStale", mock.Anything, mock.Anything, mock.Anything)
	mockJobRepo.AssertExpectations(t)
	mockAcctRepo.AssertExpectations(t)
}

func TestRunSyncPersistenceStageFailureDoesNotMarkStale(t *testing.T) {
	ctx := context.Background()
	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockJobRepo := new(mocks.MockSyncJobRepository)
	mockAssetRepo := new(mocks.MockAssetRepository)

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return &stubCollector{result: &collector.AssetResult{}}, "eu-west-2", nil
	}
	svc.SetConfigService(NewConfigService(&stubConfigRepoInternal{
		cfg: &caasmmodel.Config{
			Data: coremodel.JSONB{
				"collection_scope": map[string]interface{}{
					"ip_collection_scope": "public_only",
				},
			},
		},
	}))

	job := &caasmmodel.SyncJob{
		Base:     coremodel.Base{ID: 43},
		Metadata: coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 9},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}

	mockJobRepo.On("UpdateFields", ctx, uint64(43), mock.Anything).Return(nil)
	allowResumeCursorLookup(mockJobRepo, ctx, 9)
	mockAcctRepo.On("Update", ctx, mock.Anything).Return(nil).Once()
	mockAssetRepo.On("DeletePrivateIPAssetsForCloudAccount", mock.Anything, uint64(9)).Return(errors.New("prune failed")).Once()

	svc.runSync(ctx, job, account)

	assert.Equal(t, caasmmodel.SyncJobFailed, job.Status)
	assert.Contains(t, job.ErrorMessage, "prune failed")
	assert.NotNil(t, job.FinishedAt)
	assert.Nil(t, account.LastSyncAt)
	assert.Contains(t, account.ErrorMessage, "prune failed")
	assert.Equal(t, true, job.Metadata["collection_completed"])
	assert.Equal(t, false, job.Metadata["persist_completed"])
	assert.Equal(t, false, job.Metadata["stale_mark_completed"])
	assert.Equal(t, "", job.Metadata["stale_mark_error"])
	mockAssetRepo.AssertNotCalled(t, "SoftDeleteStale", mock.Anything, mock.Anything, mock.Anything)
	mockJobRepo.AssertExpectations(t)
	mockAcctRepo.AssertExpectations(t)
	mockAssetRepo.AssertExpectations(t)
}

func TestRunSyncPersistQueryCancellationRetriesAndSucceeds(t *testing.T) {
	ctx := context.Background()
	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockJobRepo := new(mocks.MockSyncJobRepository)
	mockAssetRepo := new(mocks.MockAssetRepository)

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return &stubCollector{result: &collector.AssetResult{
			Assets: []caasmmodel.Asset{{
				NativeID:  "persist-asset-1",
				Name:      "persist-asset-1",
				AssetType: caasmmodel.AssetTypeNetwork,
				Service:   "EC2",
				Status:    caasmmodel.AssetStatusActive,
			}},
		}}, "eu-west-2", nil
	}

	job := &caasmmodel.SyncJob{
		Base:     coremodel.Base{ID: 431},
		Metadata: coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 91},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}

	mockJobRepo.On("UpdateFields", mock.Anything, uint64(431), mock.Anything).Return(nil)
	allowResumeCursorLookup(mockJobRepo, ctx, 91)
	mockAcctRepo.On("Update", ctx, mock.Anything).Return(nil).Once()
	mockAssetRepo.On("FindMinimalByNativeIDs", mock.Anything, uint64(91), []string{"persist-asset-1"}).Return(nil, context.Canceled).Once()
	mockAssetRepo.On("FindMinimalByNativeIDs", mock.Anything, uint64(91), []string{"persist-asset-1"}).Return([]repository.AssetMinimal{}, nil).Once()
	mockAssetRepo.On("SaveAssetSnapshot", mock.Anything, mock.MatchedBy(func(asset *caasmmodel.Asset) bool {
		return asset.NativeID == "persist-asset-1"
	})).Run(func(args mock.Arguments) {
		asset := args.Get(1).(*caasmmodel.Asset)
		asset.ID = 101
	}).Return(nil).Once()
	mockAssetRepo.On("SoftDeleteStale", ctx, uint64(91), mock.Anything).Return(nil).Once()

	svc.runSync(ctx, job, account)

	assert.Equal(t, caasmmodel.SyncJobSuccess, job.Status)
	assert.Equal(t, true, job.Metadata["persist_completed"])
	assert.Equal(t, true, job.Metadata["stale_mark_completed"])
	mockAssetRepo.AssertExpectations(t)
	mockJobRepo.AssertExpectations(t)
	mockAcctRepo.AssertExpectations(t)
}

func TestRunSyncPersistQueryCancellationFailsAfterRetryBudgetExhausted(t *testing.T) {
	ctx := context.Background()
	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockJobRepo := new(mocks.MockSyncJobRepository)
	mockAssetRepo := new(mocks.MockAssetRepository)

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return &stubCollector{result: &collector.AssetResult{
			Assets: []caasmmodel.Asset{{
				NativeID:  "persist-asset-1",
				Name:      "persist-asset-1",
				AssetType: caasmmodel.AssetTypeNetwork,
				Service:   "EC2",
				Status:    caasmmodel.AssetStatusActive,
			}},
		}}, "eu-west-2", nil
	}

	job := &caasmmodel.SyncJob{
		Base:     coremodel.Base{ID: 432},
		Metadata: coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 92},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}

	mockJobRepo.On("UpdateFields", mock.Anything, uint64(432), mock.Anything).Return(nil)
	allowResumeCursorLookup(mockJobRepo, ctx, 92)
	mockAcctRepo.On("Update", ctx, mock.Anything).Return(nil).Once()
	mockAssetRepo.On("FindMinimalByNativeIDs", mock.Anything, uint64(92), []string{"persist-asset-1"}).Return(nil, context.Canceled).Once()
	mockAssetRepo.On("FindMinimalByNativeIDs", mock.Anything, uint64(92), []string{"persist-asset-1"}).Return(nil, context.Canceled).Once()
	mockAssetRepo.On("FindMinimalByNativeIDs", mock.Anything, uint64(92), []string{"persist-asset-1"}).Return(nil, context.Canceled).Once()

	svc.runSync(ctx, job, account)

	assert.Equal(t, caasmmodel.SyncJobFailed, job.Status)
	assert.Equal(t, "Persist stage database query was canceled before snapshot completion", job.ErrorMessage)
	assert.Equal(t, "persist", job.Metadata["failed_phase"])
	assert.Equal(t, "persist_error", job.Metadata["termination_reason"])
	assert.Equal(t, "db_query_canceled", job.Metadata["cancellation_source"])
	assert.Equal(t, 1, job.Metadata["persist_asset_count"])
	assert.Equal(t, 1, job.Metadata["persist_prefetch_native_id_count"])
	assert.Equal(t, 1, job.Metadata["persist_prefetch_chunk_count"])
	assert.Equal(t, 0, job.Metadata["persist_prefetch_hit_count"])
	assert.Equal(t, 1, job.Metadata["persist_prefetch_miss_count"])
	mockAssetRepo.AssertNotCalled(t, "SoftDeleteStale", mock.Anything, mock.Anything, mock.Anything)
	mockAssetRepo.AssertExpectations(t)
	mockJobRepo.AssertExpectations(t)
	mockAcctRepo.AssertExpectations(t)
}

func TestRunSyncPartialFailuresBecomePartialSuccessAndSkipStaleMarking(t *testing.T) {
	ctx := context.Background()
	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockJobRepo := new(mocks.MockSyncJobRepository)
	mockAssetRepo := new(mocks.MockAssetRepository)
	eventRepo := &stubSyncJobEventRepo{}

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.events = eventRepo
	svc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return &stubCollector{result: &collector.AssetResult{
			PartialFailures: []collector.CollectionIssue{
				{Scope: "service", Target: "eu-west-2:ec2", Message: "throttled"},
			},
		}}, "eu-west-2", nil
	}

	job := &caasmmodel.SyncJob{
		Base:     coremodel.Base{ID: 44},
		Metadata: coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 10},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}

	mockJobRepo.On("UpdateFields", ctx, uint64(44), mock.Anything).Return(nil)
	allowResumeCursorLookup(mockJobRepo, ctx, 10)
	mockAcctRepo.On("Update", ctx, mock.Anything).Return(nil).Once()

	svc.runSync(ctx, job, account)

	assert.Equal(t, caasmmodel.SyncJobPartial, job.Status)
	assert.Contains(t, job.ErrorMessage, "partial failures")
	assert.NotNil(t, job.FinishedAt)
	assert.Nil(t, account.LastSyncAt)
	assert.Contains(t, account.ErrorMessage, "partial failures")
	assert.Equal(t, true, job.Metadata["collection_completed"])
	assert.Equal(t, true, job.Metadata["persist_completed"])
	assert.Equal(t, false, job.Metadata["stale_mark_completed"])
	assert.Equal(t, "partial_failures", job.Metadata["stale_mark_skipped_reason"])
	assert.Equal(t, 1, job.Metadata["partial_failure_count"])
	assert.Equal(t, true, job.Metadata["completed_with_warnings"])
	assert.Equal(t, "aws_throttle", job.Metadata["slowdown_source"])
	assert.Equal(t, "partial_failure", job.Metadata["throttle_state"])
	assert.Equal(t, "eu-west-2:ec2", job.Metadata["throttle_target"])
	require.Len(t, eventRepo.events, 5)
	assert.Equal(t, "sync completed with 1 partial failures", eventRepo.events[4].Message)
	assert.Equal(t, 1, eventRepo.events[4].Metadata["partial_failure_count"])
	assert.Len(t, eventRepo.events[4].Metadata["partial_failures"], 1)
	mockAssetRepo.AssertNotCalled(t, "SoftDeleteStale", mock.Anything, mock.Anything, mock.Anything)
	mockJobRepo.AssertExpectations(t)
	mockAcctRepo.AssertExpectations(t)
	mockAssetRepo.AssertExpectations(t)
}

func TestRunSyncStaleMarkFailureKeepsSuccessAndRecordsStageError(t *testing.T) {
	ctx := context.Background()
	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockJobRepo := new(mocks.MockSyncJobRepository)
	mockAssetRepo := new(mocks.MockAssetRepository)
	eventRepo := &stubSyncJobEventRepo{}

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.events = eventRepo
	svc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return &stubCollector{result: &collector.AssetResult{}}, "eu-west-2", nil
	}

	job := &caasmmodel.SyncJob{
		Base:     coremodel.Base{ID: 46},
		Metadata: coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 12},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}

	mockJobRepo.On("UpdateFields", ctx, uint64(46), mock.Anything).Return(nil)
	allowResumeCursorLookup(mockJobRepo, ctx, 12)
	mockAcctRepo.On("Update", ctx, mock.Anything).Return(nil).Once()
	mockAssetRepo.On("SoftDeleteStale", ctx, uint64(12), mock.Anything).Return(errors.New("stale mark failed")).Once()

	svc.runSync(ctx, job, account)

	assert.Equal(t, caasmmodel.SyncJobSuccess, job.Status)
	assert.Equal(t, "", job.ErrorMessage)
	assert.NotNil(t, job.FinishedAt)
	assert.NotNil(t, account.LastSyncAt)
	assert.Equal(t, "", account.ErrorMessage)
	assert.Equal(t, true, job.Metadata["collection_completed"])
	assert.Equal(t, true, job.Metadata["persist_completed"])
	assert.Equal(t, false, job.Metadata["stale_mark_completed"])
	assert.Equal(t, "stale mark failed", job.Metadata["stale_mark_error"])
	foundStaleMarkWarning := false
	for _, event := range eventRepo.events {
		if event == nil {
			continue
		}
		if string(event.Type) == "warning" && event.Message == "Stale marking failed" {
			foundStaleMarkWarning = true
			break
		}
	}
	assert.True(t, foundStaleMarkWarning)
	mockJobRepo.AssertExpectations(t)
	mockAcctRepo.AssertExpectations(t)
	mockAssetRepo.AssertExpectations(t)
}

func TestRunSyncInterruptedBeforePersistDoesNotMarkStale(t *testing.T) {
	ctx := context.Background()
	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockJobRepo := new(mocks.MockSyncJobRepository)
	mockAssetRepo := new(mocks.MockAssetRepository)

	configRepo := &sequencedConfigRepo{
		cfgs: []*caasmmodel.Config{
			{Data: coremodel.JSONB{}},
			{Data: coremodel.JSONB{}},
			{Data: coremodel.JSONB{}},
			{Data: coremodel.JSONB{"sync_settings": map[string]interface{}{"kill_switch_enabled": true}}},
		},
	}

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.SetConfigService(NewConfigService(configRepo))
	svc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return &stubCollector{result: &collector.AssetResult{}}, "eu-west-2", nil
	}

	job := &caasmmodel.SyncJob{
		Base:     coremodel.Base{ID: 47},
		Metadata: coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 13},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}

	mockJobRepo.On("UpdateFields", ctx, uint64(47), mock.Anything).Return(nil)
	allowResumeCursorLookup(mockJobRepo, ctx, 13)

	svc.runSync(ctx, job, account)

	assert.Equal(t, caasmmodel.SyncJobFailed, job.Status)
	assert.Contains(t, job.ErrorMessage, "kill switch")
	assert.NotNil(t, job.FinishedAt)
	assert.Equal(t, true, job.Metadata["collection_completed"])
	assert.Equal(t, false, job.Metadata["persist_completed"])
	assert.Equal(t, false, job.Metadata["stale_mark_completed"])
	assert.Equal(t, "sync_error", job.Metadata["stale_mark_skipped_reason"])
	mockAssetRepo.AssertNotCalled(t, "SoftDeleteStale", mock.Anything, mock.Anything, mock.Anything)
	mockAcctRepo.AssertNotCalled(t, "Update", mock.Anything, mock.Anything)
	mockJobRepo.AssertExpectations(t)
}

func TestRunSyncRecordsCoordinationGuardMetadataWhenAccountAlreadyLocked(t *testing.T) {
	ctx := context.Background()
	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockJobRepo := new(mocks.MockSyncJobRepository)
	mockAssetRepo := new(mocks.MockAssetRepository)

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)

	job := &caasmmodel.SyncJob{
		Base:           coremodel.Base{ID: 45},
		CloudAccountID: func() *uint64 { v := uint64(11); return &v }(),
		Metadata:       coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 11},
		Provider: caasmmodel.CloudProviderAWS,
	}

	assert.True(t, svc.tryLockAccount(account.ID))
	defer svc.unlockAccount(account.ID)

	mockJobRepo.On("UpdateFields", ctx, uint64(45), mock.Anything).Return(nil)

	svc.runSync(ctx, job, account)

	assert.Equal(t, caasmmodel.SyncJobFailed, job.Status)
	assert.Contains(t, job.ErrorMessage, "already running")
	assert.Equal(t, "coordination_guard", job.Metadata["slowdown_source"])
	assert.Equal(t, "account_overlap", job.Metadata["coordination_state"])
	assert.Equal(t, "in_process_lock", job.Metadata["runtime_guard_scope"])
	mockJobRepo.AssertCalled(t, "UpdateFields", ctx, uint64(45), mock.MatchedBy(func(fields map[string]interface{}) bool {
		metadata, ok := fields["metadata"].(coremodel.JSONB)
		if !ok {
			return false
		}
		return fields["status"] == caasmmodel.SyncJobFailed &&
			fields["error_message"] == "another sync is already running for this account" &&
			metadata["slowdown_source"] == "coordination_guard" &&
			metadata["coordination_state"] == "account_overlap" &&
			metadata["runtime_guard_scope"] == "in_process_lock"
	}))
	mockJobRepo.AssertExpectations(t)
}

func TestInvokeLambdaSimulatedCancelStopsServerlessExecution(t *testing.T) {
	ctx := context.Background()
	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockJobRepo := new(mocks.MockSyncJobRepository)
	mockAssetRepo := new(mocks.MockAssetRepository)
	eventRepo := &stubSyncJobEventRepo{}

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.events = eventRepo
	blocker := &blockingCollector{started: make(chan struct{})}
	svc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return blocker, "eu-west-1", nil
	}

	jobID := uint64(88)
	accountID := uint64(21)
	job := &caasmmodel.SyncJob{
		Base:           coremodel.Base{ID: jobID},
		CloudAccountID: &accountID,
		Metadata:       coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: accountID},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}
	payload := &queue.SyncJobPayload{
		SyncJobID:      &jobID,
		CloudAccountID: accountID,
		Priority:       queue.PriorityNormal,
	}

	mockJobRepo.On("FindByID", mock.Anything, jobID).Return(job, nil)
	mockJobRepo.On("UpdateFields", mock.Anything, jobID, mock.Anything).Return(nil)
	mockJobRepo.On("FindLatestResumableJobForAccount", mock.Anything, accountID, jobID, "").Return((*caasmmodel.SyncJob)(nil), nil).Maybe()
	mockAcctRepo.On("FindByID", mock.Anything, accountID).Return(account, nil)
	mockAcctRepo.On("Update", mock.Anything, mock.Anything).Return(nil).Once()

	errCh := make(chan error, 1)
	go func() {
		errCh <- svc.invokeLambdaSimulated(ctx, "local-lightweight-worker", payload)
	}()

	select {
	case <-blocker.started:
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for blocking collector to start")
	}

	assert.True(t, svc.CancelSyncJob(jobID))

	select {
	case err := <-errCh:
		assert.NoError(t, err)
	case <-time.After(5 * time.Second):
		t.Fatal("timed out waiting for serverless execution to stop")
	}

	assert.Equal(t, caasmmodel.SyncJobFailed, job.Status)
	assert.Equal(t, "Job interrupted by external cancellation", job.ErrorMessage)
	assert.Equal(t, false, job.Metadata["collection_completed"])
	assert.Equal(t, false, job.Metadata["persist_completed"])
	assert.Equal(t, false, job.Metadata["stale_mark_completed"])
	assert.Equal(t, "sync_error", job.Metadata["stale_mark_skipped_reason"])
	assert.Equal(t, "external_cancellation", job.Metadata["termination_reason"])
	assert.Equal(t, "parent_context_canceled", job.Metadata["cancellation_source"])
	mockAssetRepo.AssertNotCalled(t, "SoftDeleteStale", mock.Anything, mock.Anything, mock.Anything)

	mockJobRepo.AssertExpectations(t)
	mockAcctRepo.AssertExpectations(t)
}

func eventMessages(events []*caasmmodel.SyncJobEvent) []string {
	messages := make([]string, 0, len(events))
	for _, event := range events {
		if event == nil {
			continue
		}
		messages = append(messages, event.Message)
	}
	return messages
}

func TestRunSyncChunkedPersistAdvancesCheckpointOnlyAfterSuccessfulChunk(t *testing.T) {
	ctx := context.Background()
	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockJobRepo := new(mocks.MockSyncJobRepository)
	mockAssetRepo := new(mocks.MockAssetRepository)
	configRepo := &stubConfigRepoInternal{cfg: &caasmmodel.Config{Data: coremodel.JSONB{}}}

	interruptedCtx, cancelInterrupted := context.WithCancel(ctx)
	cancelInterrupted()
	chunkCollector := &interruptedChunkCollector{
		first: &collector.AssetChunk{
			Result: collector.AssetResult{
				Assets: []caasmmodel.Asset{{NativeID: "asset-1", Name: "first"}},
			},
			Scope:    "region",
			Target:   "eu-west-1",
			Sequence: 1,
			Checkpoint: collector.ChunkCheckpoint{
				Region:   "eu-west-1",
				Eligible: true,
			},
		},
		second: &collector.AssetChunk{
			Result: collector.AssetResult{
				Assets: []caasmmodel.Asset{{NativeID: "asset-2", Name: "second"}},
			},
			Scope:    "region",
			Target:   "eu-west-2",
			Sequence: 2,
			Checkpoint: collector.ChunkCheckpoint{
				Region:   "eu-west-2",
				Eligible: true,
			},
		},
		secondCtx: interruptedCtx,
	}

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.SetConfigService(NewConfigService(configRepo))
	svc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return chunkCollector, "eu-west-1", nil
	}

	job := &caasmmodel.SyncJob{
		Base:     coremodel.Base{ID: 60},
		Metadata: coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 20},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}

	mockJobRepo.On("FindLatestResumableJobForAccount", ctx, uint64(20), uint64(60), "").Return((*caasmmodel.SyncJob)(nil), nil).Once()
	mockJobRepo.On("FindByID", mock.Anything, uint64(60)).Return(&caasmmodel.SyncJob{}, nil).Maybe()
	mockJobRepo.On("UpdateFields", mock.Anything, uint64(60), mock.Anything).Return(nil)
	mockAcctRepo.On("Update", ctx, mock.Anything).Return(nil).Once()
	mockAssetRepo.On("FindMinimalByNativeIDs", mock.Anything, uint64(20), []string{"asset-1"}).Return([]repository.AssetMinimal{}, nil).Once()
	mockAssetRepo.On("FindMinimalByNativeIDs", mock.Anything, uint64(20), []string{"asset-2"}).Return([]repository.AssetMinimal{}, nil).Once()
	mockAssetRepo.On("SaveAssetSnapshot", mock.Anything, mock.MatchedBy(func(asset *caasmmodel.Asset) bool {
		return asset.NativeID == "asset-1"
	})).Return(nil).Once()
	mockAssetRepo.On("SaveAssetSnapshot", mock.Anything, mock.MatchedBy(func(asset *caasmmodel.Asset) bool {
		return asset.NativeID == "asset-2"
	})).Return(context.Canceled).Times(3)

	svc.runSync(ctx, job, account)

	assert.Equal(t, caasmmodel.SyncJobFailed, job.Status)
	assert.Equal(t, "Persist stage database query was canceled before snapshot completion", job.ErrorMessage)
	assert.Equal(t, []string{"eu-west-1"}, resumeCursorFromMetadata(job.Metadata).CompletedRegions)
	assert.Equal(t, false, job.Metadata["stale_mark_completed"])
	assert.Equal(t, "persist", job.Metadata["failed_phase"])
	assert.Equal(t, "persist_error", job.Metadata["termination_reason"])
	assert.Equal(t, "db_query_canceled", job.Metadata["cancellation_source"])
	assert.Equal(t, "persist_error", job.Metadata["stale_mark_skipped_reason"])
	mockAssetRepo.AssertNotCalled(t, "SoftDeleteStale", mock.Anything, mock.Anything, mock.Anything)
	mockJobRepo.AssertExpectations(t)
	mockAcctRepo.AssertExpectations(t)
	mockAssetRepo.AssertExpectations(t)
}

func TestRunSyncChunkedResumeReusesWindowStartForStaleMarking(t *testing.T) {
	ctx := context.Background()
	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockJobRepo := new(mocks.MockSyncJobRepository)
	mockJobRepo.On("FindByID", mock.Anything, mock.AnythingOfType("uint64")).Return(&caasmmodel.SyncJob{}, nil).Maybe()
	mockAssetRepo := new(mocks.MockAssetRepository)

	resumeWindow := time.Date(2026, 4, 9, 12, 0, 0, 0, time.UTC)
	chunkCollector := &chunkAwareStubCollector{
		chunks: []*collector.AssetChunk{
			{
				Result: collector.AssetResult{
					Assets: []caasmmodel.Asset{{NativeID: "asset-3", Name: "third"}},
				},
				Scope:    "region",
				Target:   "eu-west-2",
				Sequence: 1,
				Checkpoint: collector.ChunkCheckpoint{
					Region:   "eu-west-2",
					Eligible: true,
				},
			},
		},
		summary: &collector.CollectionSummary{},
	}

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return chunkCollector, "eu-west-2", nil
	}

	job := &caasmmodel.SyncJob{
		Base:     coremodel.Base{ID: 61},
		Metadata: coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 21},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}
	priorCursor := collector.ResumeCursor{
		WindowStartedAt:  resumeWindow.Format(time.RFC3339Nano),
		CompletedRegions: []string{"eu-west-1"},
	}
	priorJob := caasmmodel.SyncJob{
		Base:   coremodel.Base{ID: 600},
		Status: caasmmodel.SyncJobPartial,
		Metadata: coremodel.JSONB{
			"resume_cursor": priorCursor,
		},
	}

	mockJobRepo.On("FindLatestResumableJobForAccount", ctx, uint64(21), uint64(61), "").Return(&priorJob, nil).Once()
	mockJobRepo.On("UpdateFields", mock.Anything, uint64(61), mock.Anything).Return(nil)
	mockAcctRepo.On("Update", ctx, mock.Anything).Return(nil).Once()
	mockAssetRepo.On("FindMinimalByNativeIDs", mock.Anything, uint64(21), []string{"asset-3"}).Return([]repository.AssetMinimal{}, nil).Once()
	mockAssetRepo.On("SaveAssetSnapshot", mock.Anything, mock.MatchedBy(func(asset *caasmmodel.Asset) bool {
		return asset.NativeID == "asset-3"
	})).Return(nil).Once()
	mockAssetRepo.On("SoftDeleteStale", ctx, uint64(21), mock.MatchedBy(func(cutoff time.Time) bool {
		return cutoff.Equal(resumeWindow)
	})).Return(nil).Once()

	svc.runSync(ctx, job, account)

	if assert.NotNil(t, chunkCollector.resume) {
		assert.Equal(t, priorCursor, *chunkCollector.resume)
	}
	assert.Equal(t, caasmmodel.SyncJobSuccess, job.Status)
	assert.Equal(t, []string{"eu-west-1", "eu-west-2"}, resumeCursorFromMetadata(job.Metadata).CompletedRegions)
	assert.Equal(t, true, job.Metadata["stale_mark_completed"])
	mockJobRepo.AssertExpectations(t)
	mockAcctRepo.AssertExpectations(t)
	mockAssetRepo.AssertExpectations(t)
}

type sequencedConfigRepo struct {
	mu   sync.Mutex
	cfgs []*caasmmodel.Config
	idx  int
}

func (r *sequencedConfigRepo) Get(ctx context.Context) (*caasmmodel.Config, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if len(r.cfgs) == 0 {
		return &caasmmodel.Config{}, nil
	}
	if r.idx >= len(r.cfgs) {
		return r.cfgs[len(r.cfgs)-1], nil
	}
	cfg := r.cfgs[r.idx]
	r.idx++
	return cfg, nil
}

func (r *sequencedConfigRepo) Upsert(ctx context.Context, data caasmmodel.Config) error {
	return nil
}

func TestRecordJobEventPersistsCoordinationGuardMetadataToJobRow(t *testing.T) {
	ctx := context.Background()
	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockJobRepo := new(mocks.MockSyncJobRepository)
	eventRepo := &stubSyncJobEventRepo{}

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, nil)
	svc.SetEventRepository(eventRepo)

	job := &caasmmodel.SyncJob{
		Base: coremodel.Base{ID: 88},
		Metadata: coremodel.JSONB{
			"account_name": "acct",
			"provider":     "aws",
		},
	}

	mockJobRepo.On("FindByID", ctx, uint64(88)).Return(job, nil)
	mockJobRepo.On("UpdateFields", ctx, uint64(88), mock.MatchedBy(func(fields map[string]interface{}) bool {
		rawMeta, ok := fields["metadata"].(coremodel.JSONB)
		if !ok {
			return false
		}
		return rawMeta["slowdown_source"] == "coordination_guard" &&
			rawMeta["coordination_state"] == "account_overlap" &&
			rawMeta["runtime_guard_scope"] == "in_process_lock" &&
			rawMeta["reason"] == "account_overlap" &&
			rawMeta["retry_after_sec"] == 30 &&
			rawMeta["consumer"] == "worker-1"
	})).Return(nil)

	svc.recordJobEvent(ctx, 88, "warning", "Job deferred by runtime coordination guard", map[string]interface{}{
		"slowdown_source":     "coordination_guard",
		"coordination_state":  "account_overlap",
		"runtime_guard_scope": "in_process_lock",
		"reason":              "account_overlap",
		"retry_after_sec":     30,
		"consumer":            "worker-1",
	})

	assert.Len(t, eventRepo.events, 1)
	assert.Equal(t, "warning", string(eventRepo.events[0].Type))
	assert.Equal(t, "Job deferred by runtime coordination guard", eventRepo.events[0].Message)
	assert.Equal(t, "coordination_guard", eventRepo.events[0].Metadata["slowdown_source"])
	mockJobRepo.AssertExpectations(t)
}

func TestPersistSyncPayloadInjectsCloudAccountIDForFallbackAssets(t *testing.T) {
	ctx := context.Background()

	mockAssetRepo := new(mocks.MockAssetRepository)
	svc := NewCloudAccountService(nil, nil, mockAssetRepo)

	payload := &caasmmodel.IngestPayload{
		CloudAccountID: 42,
		IPAssets: []caasmmodel.IPAsset{
			{
				Asset: caasmmodel.Asset{
					NativeID: "  fallback-asset  ",
					Name:     "fallback",
				},
				IPAddress: "10.0.0.2",
			},
		},
	}

	mockAssetRepo.On("FindMinimalByNativeIDs", ctx, uint64(42), []string{"fallback-asset"}).Return([]repository.AssetMinimal{}, nil).Once()
	mockAssetRepo.On("SaveAssetSnapshot", ctx, mock.MatchedBy(func(asset *caasmmodel.Asset) bool {
		return asset.CloudAccountID != nil &&
			*asset.CloudAccountID == 42 &&
			asset.NativeID == "fallback-asset" &&
			asset.Name == "fallback"
	})).Run(func(args mock.Arguments) {
		asset := args.Get(1).(*caasmmodel.Asset)
		asset.ID = 77
	}).Return(nil).Once()
	mockAssetRepo.On("SaveIPAsset", ctx, mock.MatchedBy(func(ipAsset *caasmmodel.IPAsset) bool {
		return ipAsset.AssetID == 77 &&
			ipAsset.IPAddress == "10.0.0.2" &&
			ipAsset.Asset.ID == 0 &&
			ipAsset.Asset.NativeID == ""
	})).Return(nil).Once()

	err := svc.persistSyncPayload(ctx, nil, payload)
	assert.NoError(t, err)
	mockAssetRepo.AssertExpectations(t)
}

func TestPersistSyncPayloadResolvesReferenceOnlyDomainAssetWithoutSavingEmptyAsset(t *testing.T) {
	ctx := context.Background()

	mockAssetRepo := new(mocks.MockAssetRepository)
	svc := NewCloudAccountService(nil, nil, mockAssetRepo)

	payload := &caasmmodel.IngestPayload{
		CloudAccountID: 42,
		DomainAssets: []caasmmodel.DomainAsset{
			{
				Asset: caasmmodel.Asset{
					NativeID: "zone-root",
				},
				DomainName:  "example.com",
				DomainClass: "hosted_zone",
				Registrar:   "route53",
			},
		},
	}

	mockAssetRepo.On("FindMinimalByNativeIDs", ctx, uint64(42), []string{"zone-root"}).Return([]repository.AssetMinimal{{
		ID:       101,
		NativeID: "zone-root",
	}}, nil).Once()
	mockAssetRepo.On("SaveDomainAssetSnapshot", ctx, mock.MatchedBy(func(domainAsset *caasmmodel.DomainAsset) bool {
		return domainAsset.AssetID == 101 &&
			domainAsset.DomainName == "example.com" &&
			domainAsset.DomainClass == "hosted_zone" &&
			domainAsset.Asset.ID == 0 &&
			domainAsset.Asset.NativeID == "" &&
			domainAsset.Asset.Service == ""
	})).Return(nil).Once()

	err := svc.persistSyncPayload(ctx, nil, payload)
	assert.NoError(t, err)
	mockAssetRepo.AssertNotCalled(t, "FindByNativeID", mock.Anything, mock.Anything, mock.Anything)
	mockAssetRepo.AssertNotCalled(t, "Save", mock.Anything, mock.Anything)
	mockAssetRepo.AssertExpectations(t)
}

func TestPersistSyncPayloadDoesNotOverwriteAssetWithServiceOnlyDomainReference(t *testing.T) {
	ctx := context.Background()

	mockAssetRepo := new(mocks.MockAssetRepository)
	svc := NewCloudAccountService(nil, nil, mockAssetRepo)

	payload := &caasmmodel.IngestPayload{
		CloudAccountID: 42,
		Assets: []caasmmodel.Asset{
			{
				NativeID:  "lb-arn",
				Name:      "app-lb",
				AssetType: caasmmodel.AssetTypeNetwork,
				Service:   "ELB",
			},
		},
		DomainAssets: []caasmmodel.DomainAsset{
			{
				Asset: caasmmodel.Asset{
					NativeID: "lb-arn",
					Service:  "ELB",
				},
				DomainName:  "app-lb.example.elb.amazonaws.com",
				DomainClass: caasmmodel.DomainClassServiceHostname,
				Registrar:   "aws",
			},
		},
	}

	mockAssetRepo.On("FindMinimalByNativeIDs", ctx, uint64(42), []string{"lb-arn"}).Return([]repository.AssetMinimal{}, nil).Once()
	mockAssetRepo.On("SaveAssetSnapshot", ctx, mock.MatchedBy(func(asset *caasmmodel.Asset) bool {
		return asset.NativeID == "lb-arn" &&
			asset.Name == "app-lb" &&
			asset.AssetType == caasmmodel.AssetTypeNetwork
	})).Run(func(args mock.Arguments) {
		asset := args.Get(1).(*caasmmodel.Asset)
		asset.ID = 202
	}).Return(nil).Once()
	mockAssetRepo.On("SaveDomainAssetSnapshot", ctx, mock.MatchedBy(func(domainAsset *caasmmodel.DomainAsset) bool {
		return domainAsset.AssetID == 202 &&
			domainAsset.DomainName == "app-lb.example.elb.amazonaws.com" &&
			domainAsset.Asset.ID == 0 &&
			domainAsset.Asset.NativeID == "" &&
			domainAsset.Asset.Service == ""
	})).Return(nil).Once()

	err := svc.persistSyncPayload(ctx, nil, payload)
	assert.NoError(t, err)
	mockAssetRepo.AssertExpectations(t)
}

func TestPersistSyncPayloadFailsClosedWhenReferenceOnlyDomainAssetMissingBackingAsset(t *testing.T) {
	ctx := context.Background()

	mockAssetRepo := new(mocks.MockAssetRepository)
	svc := NewCloudAccountService(nil, nil, mockAssetRepo)

	payload := &caasmmodel.IngestPayload{
		CloudAccountID: 42,
		DomainAssets: []caasmmodel.DomainAsset{
			{
				Asset: caasmmodel.Asset{
					NativeID: "zone-root",
				},
				DomainName:  "missing.example.com",
				DomainClass: "hosted_zone",
			},
		},
	}

	mockAssetRepo.On("FindMinimalByNativeIDs", ctx, uint64(42), []string{"zone-root"}).Return([]repository.AssetMinimal{}, nil).Once()

	err := svc.persistSyncPayload(ctx, nil, payload)
	require.Error(t, err)
	require.Contains(t, err.Error(), "resolve reference-only asset")
	mockAssetRepo.AssertNotCalled(t, "FindByNativeID", mock.Anything, mock.Anything, mock.Anything)
	mockAssetRepo.AssertNotCalled(t, "Save", mock.Anything, mock.Anything)
	mockAssetRepo.AssertNotCalled(t, "SaveDomainAssetSnapshot", mock.Anything, mock.Anything)
	mockAssetRepo.AssertExpectations(t)
}

func TestPersistSyncPayloadUsesCanonicalDNSRecordIdentityForInheritance(t *testing.T) {
	ctx := context.Background()

	mockAssetRepo := new(mocks.MockAssetRepository)
	dnsRepo := &stubDNSRecordRepoInternal{
		existing: []caasmmodel.DNSRecord{
			{
				Base:              coremodel.Base{ID: 11},
				DomainAssetID:     500,
				Name:              "www.example.com",
				Type:              "A",
				TTL:               300,
				Values:            coremodel.StringArray{"10.0.0.1"},
				SetIdentifier:     func() *string { v := "blue"; return &v }(),
				OwnerEmail:        "blue@example.com",
				OwnerDepartmentID: "dept-blue",
				OwnerEmployeeID:   func() *uint64 { v := uint64(101); return &v }(),
				IgnoreChanges:     true,
				SuppressPattern:   "blue.*",
				LastChangeAt:      func() *time.Time { v := time.Date(2026, 4, 1, 12, 0, 0, 0, time.UTC); return &v }(),
				LastChangeFrom:    coremodel.StringArray{"old-blue"},
				LastChangeTo:      coremodel.StringArray{"new-blue"},
			},
			{
				Base:              coremodel.Base{ID: 12},
				DomainAssetID:     500,
				Name:              "www.example.com",
				Type:              "A",
				TTL:               300,
				Values:            coremodel.StringArray{"10.0.0.2"},
				Failover:          "PRIMARY",
				OwnerEmail:        "primary@example.com",
				OwnerDepartmentID: "dept-primary",
				OwnerEmployeeID:   func() *uint64 { v := uint64(202); return &v }(),
				IgnoreChanges:     false,
				SuppressPattern:   "primary.*",
				LastChangeAt:      func() *time.Time { v := time.Date(2026, 4, 1, 12, 5, 0, 0, time.UTC); return &v }(),
				LastChangeFrom:    coremodel.StringArray{"old-primary"},
				LastChangeTo:      coremodel.StringArray{"new-primary"},
			},
			{
				Base:              coremodel.Base{ID: 13},
				DomainAssetID:     500,
				Name:              "www.example.com",
				Type:              "A",
				TTL:               300,
				Values:            coremodel.StringArray{"10.0.0.3"},
				Region:            "us-east-1",
				OwnerEmail:        "east@example.com",
				OwnerDepartmentID: "dept-east",
				OwnerEmployeeID:   func() *uint64 { v := uint64(303); return &v }(),
				IgnoreChanges:     true,
				SuppressPattern:   "east.*",
				LastChangeAt:      func() *time.Time { v := time.Date(2026, 4, 1, 12, 10, 0, 0, time.UTC); return &v }(),
				LastChangeFrom:    coremodel.StringArray{"old-east"},
				LastChangeTo:      coremodel.StringArray{"new-east"},
			},
		},
	}
	svc := NewCloudAccountService(nil, nil, mockAssetRepo)
	svc.SetDNSRecordRepository(dnsRepo)

	payload := &caasmmodel.IngestPayload{
		CloudAccountID: 42,
		DomainAssets: []caasmmodel.DomainAsset{
			{
				AssetID:     500,
				DomainName:  "example.com",
				DomainClass: caasmmodel.DomainClassDNSRecord,
				Registrar:   "route53",
				DNSRecords: coremodel.JSONB{
					"records": []map[string]interface{}{
						{"name": "www.example.com.", "type": "A", "set_identifier": "blue", "values": []string{"10.0.0.1"}},
						{"name": "www.example.com.", "type": "A", "set_identifier": "green", "values": []string{"10.0.0.4"}},
						{"name": "www.example.com.", "type": "A", "failover": "PRIMARY", "values": []string{"10.0.0.2"}},
						{"name": "www.example.com.", "type": "A", "failover": "SECONDARY", "values": []string{"10.0.0.5"}},
						{"name": "www.example.com.", "type": "A", "region": "us-east-1", "values": []string{"10.0.0.3"}},
						{"name": "www.example.com.", "type": "A", "region": "us-west-2", "values": []string{"10.0.0.6"}},
					},
				},
			},
		},
	}

	mockAssetRepo.On("SaveDomainAssetSnapshot", ctx, mock.MatchedBy(func(domainAsset *caasmmodel.DomainAsset) bool {
		return domainAsset.AssetID == 500 &&
			domainAsset.DomainName == "example.com" &&
			domainAsset.DomainClass == caasmmodel.DomainClassDNSRecord &&
			domainAsset.Asset.ID == 0 &&
			domainAsset.Asset.NativeID == "" &&
			domainAsset.Asset.Service == ""
	})).Run(func(args mock.Arguments) {
		domainAsset := args.Get(1).(*caasmmodel.DomainAsset)
		domainAsset.ID = 500
	}).Return(nil).Once()

	require.NoError(t, svc.persistSyncPayload(ctx, nil, payload))
	mockAssetRepo.AssertExpectations(t)
	require.Equal(t, uint64(500), dnsRepo.replacedDomainID)
	require.Len(t, dnsRepo.replacedRecords, 6)

	findRecord := func(identity dnsRecordPersistIdentity) *caasmmodel.DNSRecord {
		for i := range dnsRepo.replacedRecords {
			rec := &dnsRepo.replacedRecords[i]
			key, err := normalizePersistedDNSRecordIdentity(rec)
			require.NoError(t, err)
			if key == identity {
				return rec
			}
		}
		return nil
	}

	blueKey := dnsRecordPersistIdentity{DomainAssetID: 500, NameKey: "www.example.com", TypeKey: "A", SetIdentifierKey: "blue"}
	greenKey := dnsRecordPersistIdentity{DomainAssetID: 500, NameKey: "www.example.com", TypeKey: "A", SetIdentifierKey: "green"}
	primaryKey := dnsRecordPersistIdentity{DomainAssetID: 500, NameKey: "www.example.com", TypeKey: "A", FailoverKey: "PRIMARY"}
	secondaryKey := dnsRecordPersistIdentity{DomainAssetID: 500, NameKey: "www.example.com", TypeKey: "A", FailoverKey: "SECONDARY"}
	eastKey := dnsRecordPersistIdentity{DomainAssetID: 500, NameKey: "www.example.com", TypeKey: "A", RegionKey: "us-east-1"}
	westKey := dnsRecordPersistIdentity{DomainAssetID: 500, NameKey: "www.example.com", TypeKey: "A", RegionKey: "us-west-2"}

	blue := findRecord(blueKey)
	green := findRecord(greenKey)
	primary := findRecord(primaryKey)
	secondary := findRecord(secondaryKey)
	east := findRecord(eastKey)
	west := findRecord(westKey)

	require.NotNil(t, blue)
	require.NotNil(t, green)
	require.NotNil(t, primary)
	require.NotNil(t, secondary)
	require.NotNil(t, east)
	require.NotNil(t, west)

	assert.Equal(t, "blue@example.com", blue.OwnerEmail)
	assert.Equal(t, "dept-blue", blue.OwnerDepartmentID)
	require.NotNil(t, blue.OwnerEmployeeID)
	assert.Equal(t, uint64(101), *blue.OwnerEmployeeID)
	assert.True(t, blue.IgnoreChanges)
	assert.Equal(t, "blue.*", blue.SuppressPattern)
	require.NotNil(t, blue.LastChangeAt)
	assert.True(t, blue.LastChangeAt.Equal(time.Date(2026, 4, 1, 12, 0, 0, 0, time.UTC)))
	assert.Equal(t, coremodel.StringArray{"old-blue"}, blue.LastChangeFrom)
	assert.Equal(t, coremodel.StringArray{"new-blue"}, blue.LastChangeTo)

	assert.Equal(t, "primary@example.com", primary.OwnerEmail)
	assert.Equal(t, "dept-primary", primary.OwnerDepartmentID)
	require.NotNil(t, primary.OwnerEmployeeID)
	assert.Equal(t, uint64(202), *primary.OwnerEmployeeID)
	assert.False(t, primary.IgnoreChanges)
	assert.Equal(t, "primary.*", primary.SuppressPattern)
	require.NotNil(t, primary.LastChangeAt)
	assert.True(t, primary.LastChangeAt.Equal(time.Date(2026, 4, 1, 12, 5, 0, 0, time.UTC)))
	assert.Equal(t, coremodel.StringArray{"old-primary"}, primary.LastChangeFrom)
	assert.Equal(t, coremodel.StringArray{"new-primary"}, primary.LastChangeTo)

	assert.Equal(t, "east@example.com", east.OwnerEmail)
	assert.Equal(t, "dept-east", east.OwnerDepartmentID)
	require.NotNil(t, east.OwnerEmployeeID)
	assert.Equal(t, uint64(303), *east.OwnerEmployeeID)
	assert.True(t, east.IgnoreChanges)
	assert.Equal(t, "east.*", east.SuppressPattern)
	require.NotNil(t, east.LastChangeAt)
	assert.True(t, east.LastChangeAt.Equal(time.Date(2026, 4, 1, 12, 10, 0, 0, time.UTC)))
	assert.Equal(t, coremodel.StringArray{"old-east"}, east.LastChangeFrom)
	assert.Equal(t, coremodel.StringArray{"new-east"}, east.LastChangeTo)

	assert.Empty(t, green.OwnerEmail)
	assert.Empty(t, green.OwnerDepartmentID)
	assert.Nil(t, green.OwnerEmployeeID)
	assert.False(t, green.IgnoreChanges)
	assert.Empty(t, green.SuppressPattern)
	assert.Nil(t, green.LastChangeAt)
	assert.Empty(t, green.LastChangeFrom)
	assert.Empty(t, green.LastChangeTo)

	assert.Empty(t, secondary.OwnerEmail)
	assert.Empty(t, secondary.OwnerDepartmentID)
	assert.Nil(t, secondary.OwnerEmployeeID)
	assert.False(t, secondary.IgnoreChanges)
	assert.Empty(t, secondary.SuppressPattern)
	assert.Nil(t, secondary.LastChangeAt)
	assert.Empty(t, secondary.LastChangeFrom)
	assert.Empty(t, secondary.LastChangeTo)

	assert.Empty(t, west.OwnerEmail)
	assert.Empty(t, west.OwnerDepartmentID)
	assert.Nil(t, west.OwnerEmployeeID)
	assert.False(t, west.IgnoreChanges)
	assert.Empty(t, west.SuppressPattern)
	assert.Nil(t, west.LastChangeAt)
	assert.Empty(t, west.LastChangeFrom)
	assert.Empty(t, west.LastChangeTo)
}

func TestPersistSyncPayloadIgnoresAssetOwnerFieldsWhenSavingSnapshot(t *testing.T) {
	ctx := context.Background()

	mockAssetRepo := new(mocks.MockAssetRepository)
	svc := NewCloudAccountService(nil, nil, mockAssetRepo)

	accountID := uint64(42)
	payload := &caasmmodel.IngestPayload{
		CloudAccountID: accountID,
		Assets: []caasmmodel.Asset{
			{
				CloudAccountID:  &accountID,
				NativeID:        "asset-owner-1",
				Name:            "asset-owner-1",
				AssetType:       caasmmodel.AssetTypeNetwork,
				Service:         "EC2",
				Status:          caasmmodel.AssetStatusActive,
				OwnerEmployeeID: func() *uint64 { v := uint64(501); return &v }(),
				DepartmentID:    func() *string { v := "dept-existing"; return &v }(),
			},
		},
	}

	mockAssetRepo.On("FindMinimalByNativeIDs", ctx, accountID, []string{"asset-owner-1"}).Return([]repository.AssetMinimal{}, nil).Once()
	mockAssetRepo.On("SaveAssetSnapshot", ctx, mock.MatchedBy(func(asset *caasmmodel.Asset) bool {
		if asset == nil {
			return false
		}
		if asset.OwnerEmployeeID != nil {
			return false
		}
		return asset.DepartmentID == nil
	})).Run(func(args mock.Arguments) {
		asset := args.Get(1).(*caasmmodel.Asset)
		asset.ID = 77
	}).Return(nil).Once()

	require.NoError(t, svc.persistSyncPayload(ctx, nil, payload))
	mockAssetRepo.AssertNotCalled(t, "FindByNativeID", mock.Anything, mock.Anything, mock.Anything)
	mockAssetRepo.AssertExpectations(t)
}

func TestPersistSyncPayloadIgnoresDomainOwnerFieldsWhenSavingSnapshot(t *testing.T) {
	ctx := context.Background()

	mockAssetRepo := new(mocks.MockAssetRepository)
	svc := NewCloudAccountService(nil, nil, mockAssetRepo)

	payload := &caasmmodel.IngestPayload{
		CloudAccountID: 42,
		DomainAssets: []caasmmodel.DomainAsset{
			{
				AssetID:         101,
				DomainName:      "example.com",
				DomainClass:     "hosted_zone",
				Registrar:       "route53",
				OwnerEmployeeID: func() *uint64 { v := uint64(902); return &v }(),
				DepartmentID:    func() *string { v := "dept-domain"; return &v }(),
			},
		},
	}

	mockAssetRepo.On("SaveDomainAssetSnapshot", ctx, mock.MatchedBy(func(domainAsset *caasmmodel.DomainAsset) bool {
		if domainAsset == nil {
			return false
		}
		if domainAsset.OwnerEmployeeID != nil {
			return false
		}
		return domainAsset.DepartmentID == nil
	})).Run(func(args mock.Arguments) {
		domainAsset := args.Get(1).(*caasmmodel.DomainAsset)
		domainAsset.ID = 101
	}).Return(nil).Once()

	require.NoError(t, svc.persistSyncPayload(ctx, nil, payload))
	mockAssetRepo.AssertNotCalled(t, "SaveDomainAsset", mock.Anything, mock.Anything)
	mockAssetRepo.AssertExpectations(t)
}