package service

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"sync/atomic"
	"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/repository"
)

type chunkAwareStubCollector struct {
	resume          *collector.ResumeCursor
	chunks          []*collector.AssetChunk
	partialFailures []collector.CollectionIssue
	summary         *collector.CollectionSummary
	err             error
}

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

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

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

func (c *chunkAwareStubCollector) SetResumeCursor(cursor *collector.ResumeCursor) {
	if cursor == nil {
		c.resume = nil
		return
	}
	copyCursor := *cursor
	c.resume = &copyCursor
}

func (c *chunkAwareStubCollector) CollectChunks(ctx context.Context, sink collector.ChunkSink) (*collector.CollectionSummary, error) {
	summary := c.summary
	if summary == nil {
		summary = &collector.CollectionSummary{CollectedAt: time.Date(2026, 4, 9, 12, 0, 0, 0, time.UTC)}
	}
	for _, chunk := range c.chunks {
		if chunk == nil {
			continue
		}
		if err := sink.PersistChunk(ctx, chunk); err != nil {
			summary.PartialFailures = append(summary.PartialFailures, collector.CollectionIssue{
				Scope:   chunk.Scope,
				Target:  chunk.Target,
				Message: err.Error(),
			})
			continue
		}
		summary.ChunkCount++
	}
	summary.PartialFailures = append(summary.PartialFailures, c.partialFailures...)
	return summary, c.err
}

func TestUpdateSyncJobMetadataMergesCheckpointAndStageFlags(t *testing.T) {
	ctx := context.Background()
	mockJobRepo := new(mocks.MockSyncJobRepository)
	svc := NewCloudAccountService(nil, mockJobRepo, nil)

	cursor := collector.ResumeCursor{
		WindowStartedAt:       "2026-04-09T12:00:00Z",
		CompletedGlobal:       []string{"route53"},
		CompletedRegions:      []string{"eu-west-2"},
		Route53CompletedZones: []string{"ZTEST"},
	}
	job := &caasmmodel.SyncJob{
		Base: coremodel.Base{ID: 701},
		Metadata: coremodel.JSONB{
			"collection_completed": true,
			"resume_cursor":        resumeCursorJSONB(cursor),
		},
	}

	mockJobRepo.On("UpdateFields", mock.Anything, uint64(701), mock.Anything).Return(nil).Twice()

	svc.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
		"persist_completed": true,
	})
	svc.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
		"stale_mark_completed": true,
	})

	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, resumeCursorJSONB(cursor), job.Metadata["resume_cursor"])
	mockJobRepo.AssertExpectations(t)
}

func TestUpdateSyncJobMetadataDoesNotMutateJobWhenPersistFails(t *testing.T) {
	ctx := context.Background()
	mockJobRepo := new(mocks.MockSyncJobRepository)
	svc := NewCloudAccountService(nil, mockJobRepo, nil)

	cursor := collector.ResumeCursor{
		WindowStartedAt: "2026-04-09T12:00:00Z",
		CompletedGlobal: []string{"route53"},
	}
	job := &caasmmodel.SyncJob{
		Base: coremodel.Base{ID: 702},
		Metadata: coremodel.JSONB{
			"resume_cursor": resumeCursorJSONB(cursor),
		},
	}

	mockJobRepo.On("UpdateFields", mock.Anything, uint64(702), mock.Anything).Return(errors.New("metadata write failed")).Once()

	err := svc.updateSyncJobMetadata(ctx, job, coremodel.JSONB{
		"persist_completed": true,
		"resume_cursor": resumeCursorJSONB(collector.ResumeCursor{
			WindowStartedAt: "2026-04-09T12:00:00Z",
			CompletedGlobal: []string{"route53", "cloudfront"},
		}),
	})

	require.Error(t, err)
	assert.NotContains(t, job.Metadata, "persist_completed")
	assert.Equal(t, resumeCursorJSONB(cursor), job.Metadata["resume_cursor"])
	mockJobRepo.AssertExpectations(t)
}

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

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.SetConfigService(NewConfigService(&stubConfigRepoInternal{
		cfg: &caasmmodel.Config{
			Data: coremodel.JSONB{
				"collection_scope": map[string]interface{}{
					"ip_collection_scope":   string(caasmmodel.CollectionScopePublicOnly),
					"port_collection_scope": string(caasmmodel.CollectionScopePublicOnly),
				},
			},
		},
	}))

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

	mockJobRepo.On("UpdateFields", mock.Anything, uint64(801), mock.Anything).Return(nil)
	mockAssetRepo.On("DeletePrivateIPAssetsForCloudAccount", mock.Anything, uint64(99)).Return(nil).Once()

	sink := newChunkedSyncSink(svc, job, account, ctx, collector.ResumeCursor{}, 0, time.Date(2026, 4, 9, 12, 0, 0, 0, time.UTC))
	first := &collector.AssetChunk{
		Result:   collector.AssetResult{},
		Scope:    "route53",
		Target:   "example.com",
		Sequence: 1,
		Checkpoint: collector.ChunkCheckpoint{
			Route53ZoneID:       "Z1",
			Route53Zone:         "example.com",
			Route53ZoneComplete: true,
			Eligible:            true,
		},
	}
	second := &collector.AssetChunk{
		Result:   collector.AssetResult{},
		Scope:    "route53",
		Target:   "api.example.com",
		Sequence: 2,
		Checkpoint: collector.ChunkCheckpoint{
			Route53ZoneID:       "Z2",
			Route53Zone:         "api.example.com",
			Route53ZoneComplete: true,
			Eligible:            true,
		},
	}

	assert.NoError(t, sink.PersistChunk(ctx, first))
	assert.NoError(t, sink.PersistChunk(ctx, second))
	assert.NoError(t, sink.Wait())

	assert.Equal(t, 2, sink.chunkCount)
	assert.Equal(t, 1, job.Metadata["chunk_persist_count"])
	assert.Equal(t, 1, job.Metadata["last_chunk_sequence"])
	resumeCursor, ok := job.Metadata["resume_cursor"].(coremodel.JSONB)
	if !ok {
		t.Fatalf("resume_cursor type = %T, want coremodel.JSONB", job.Metadata["resume_cursor"])
	}
	assert.Equal(t, []string{"Z1", "Z2"}, resumeCursor["route53_completed_zones"])
	assert.Equal(t, []string(nil), resumeCursor["completed_global"])
	assert.Equal(t, []string(nil), resumeCursor["completed_regions"])
	mockAssetRepo.AssertExpectations(t)
	mockJobRepo.AssertExpectations(t)
}

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

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	job := &caasmmodel.SyncJob{
		Base:     coremodel.Base{ID: 8011},
		Metadata: coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 99},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}

	mockJobRepo.On("UpdateFields", mock.Anything, uint64(8011), mock.Anything).Return(nil).Once()
	mockJobRepo.On("UpdateFields", mock.Anything, uint64(8011), mock.Anything).Return(errors.New("metadata write failed")).Once()

	sink := newChunkedSyncSink(svc, job, account, ctx, collector.ResumeCursor{}, 0, time.Date(2026, 4, 9, 12, 0, 0, 0, time.UTC))
	chunk := &collector.AssetChunk{
		Result:   collector.AssetResult{},
		Scope:    "route53",
		Target:   "example.com",
		Sequence: 1,
		Checkpoint: collector.ChunkCheckpoint{
			Route53ZoneID:       "Z1",
			Route53Zone:         "example.com",
			Route53ZoneComplete: true,
			Eligible:            true,
		},
	}

	assert.NoError(t, sink.PersistChunk(ctx, chunk))
	err := sink.Wait()
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "persist checkpoint metadata")
	assert.NotContains(t, job.Metadata, "resume_cursor")
	mockJobRepo.AssertExpectations(t)
}

type checkpointFailingSyncJobRepo struct {
	phase2SyncJobRepo
	failZoneID                string
	failedCheckpointWrite     bool
	metadataWritesAfterFailed []coremodel.JSONB
}

func (r *checkpointFailingSyncJobRepo) UpdateFields(ctx context.Context, _ uint64, fields map[string]interface{}) error {
	r.updateFieldsCalled = true
	r.updatedFields = fields
	r.updateCtxErr = ctx.Err()
	if r.job != nil {
		if status, ok := fields["status"].(caasmmodel.SyncJobStatus); ok {
			r.job.Status = status
		}
	}
	rawMetadata, hasMetadata := fields["metadata"]
	if !hasMetadata {
		return nil
	}
	metadata, ok := rawMetadata.(coremodel.JSONB)
	if !ok {
		return nil
	}
	metadataCopy := coremodel.JSONB{}
	for key, value := range metadata {
		metadataCopy[key] = value
	}
	if r.failedCheckpointWrite {
		r.metadataWritesAfterFailed = append(r.metadataWritesAfterFailed, metadataCopy)
	}
	cursor := resumeCursorFromMetadata(metadataCopy)
	if slices.Contains(cursor.Route53CompletedZones, r.failZoneID) {
		r.failedCheckpointWrite = true
		return errors.New("checkpoint metadata write failed")
	}
	if r.job != nil {
		r.job.Metadata = metadataCopy
	}
	return nil
}

func TestRunSyncCheckpointMetadataFailureDoesNotLeakResumeCursorInTerminalMetadata(t *testing.T) {
	ctx := context.Background()
	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockAssetRepo := new(mocks.MockAssetRepository)
	job := &caasmmodel.SyncJob{
		Base:     coremodel.Base{ID: 8012},
		Metadata: coremodel.JSONB{},
	}
	jobRepo := &checkpointFailingSyncJobRepo{
		phase2SyncJobRepo: phase2SyncJobRepo{
			job: job,
		},
		failZoneID: "ZLEAK",
	}

	svc := NewCloudAccountService(mockAcctRepo, jobRepo, mockAssetRepo)
	svc.SetConfigService(NewConfigService(&stubConfigRepoInternal{
		cfg: &caasmmodel.Config{
			Data: coremodel.JSONB{
				"collection_scope": map[string]interface{}{
					"ip_collection_scope":   string(caasmmodel.CollectionScopePublicOnly),
					"port_collection_scope": string(caasmmodel.CollectionScopePublicOnly),
				},
			},
		},
	}))
	svc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return &chunkAwareStubCollector{
			chunks: []*collector.AssetChunk{
				{
					Result:   collector.AssetResult{},
					Scope:    "route53",
					Target:   "leak.example.com",
					Sequence: 1,
					Checkpoint: collector.ChunkCheckpoint{
						Route53ZoneID:       "ZLEAK",
						Route53Zone:         "leak.example.com",
						Route53ZoneComplete: true,
						Eligible:            true,
					},
				},
			},
		}, "us-east-1", nil
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 8012},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}

	mockAssetRepo.On("DeletePrivateIPAssetsForCloudAccount", mock.Anything, uint64(8012)).Return(nil).Once()
	mockAcctRepo.On("Update", mock.Anything, mock.Anything).Return(nil).Once()

	svc.runSync(ctx, job, account)

	require.True(t, jobRepo.failedCheckpointWrite)
	require.NotEmpty(t, jobRepo.metadataWritesAfterFailed)
	assert.Equal(t, caasmmodel.SyncJobFailed, job.Status)
	assert.Contains(t, job.ErrorMessage, "persist checkpoint metadata")
	assert.Equal(t, "sync_error", job.Metadata["termination_reason"])
	assert.NotContains(t, resumeCursorFromMetadata(job.Metadata).Route53CompletedZones, "ZLEAK")
	for _, metadata := range jobRepo.metadataWritesAfterFailed {
		assert.NotContains(t, resumeCursorFromMetadata(metadata).Route53CompletedZones, "ZLEAK")
	}
	mockAssetRepo.AssertExpectations(t)
	mockAcctRepo.AssertExpectations(t)
}

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

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.events = eventRepo

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

	updateCalls := 0
	mockJobRepo.On("UpdateFields", mock.Anything, uint64(802), mock.Anything).Run(func(args mock.Arguments) {
		updateCalls++
	}).Return(nil).Maybe()
	mockAssetRepo.On("SaveAssetSnapshot", mock.Anything, mock.Anything).Return(nil).Times(3)

	sink := newChunkedSyncSink(svc, job, account, ctx, collector.ResumeCursor{}, 0, time.Date(2026, 4, 9, 12, 0, 0, 0, time.UTC))
	first := &collector.AssetChunk{
		Result: collector.AssetResult{
			Assets: []caasmmodel.Asset{{NativeID: "asset-1", Name: "first"}},
		},
		Scope:    "region_service",
		Target:   "eu-west-1/ec2",
		Sequence: 1,
	}
	second := &collector.AssetChunk{
		Result: collector.AssetResult{
			Assets: []caasmmodel.Asset{{NativeID: "asset-2", Name: "second"}},
		},
		Scope:    "region_service",
		Target:   "eu-west-1/rds",
		Sequence: 2,
	}
	third := &collector.AssetChunk{
		Result: collector.AssetResult{
			Assets: []caasmmodel.Asset{{NativeID: "asset-3", Name: "third"}},
		},
		Scope:    "region_service",
		Target:   "eu-west-1/redshift",
		Sequence: 3,
		Checkpoint: collector.ChunkCheckpoint{
			Region:   "eu-west-1",
			Eligible: true,
		},
	}

	assert.NoError(t, sink.PersistChunk(ctx, first))
	assert.NoError(t, sink.PersistChunk(ctx, second))
	assert.NoError(t, sink.PersistChunk(ctx, third))
	assert.NoError(t, sink.Wait())

	assert.Equal(t, 5, updateCalls)
	assert.Equal(t, 3, sink.chunkCount)
	assert.Equal(t, 1, job.Metadata["chunk_persist_count"])
	assert.Equal(t, 1, job.Metadata["last_chunk_sequence"])
	assert.Len(t, eventRepo.events, 2)
	assert.Equal(t, "Chunk sync prune completed", eventRepo.events[0].Message)
	assert.Equal(t, "Chunk persist finished", eventRepo.events[1].Message)
	resumeCursor, ok := job.Metadata["resume_cursor"].(coremodel.JSONB)
	if !ok {
		t.Fatalf("resume_cursor type = %T, want coremodel.JSONB", job.Metadata["resume_cursor"])
	}
	assert.Equal(t, []string{"eu-west-1"}, resumeCursor["completed_regions"])
	mockAssetRepo.AssertExpectations(t)
	mockJobRepo.AssertExpectations(t)
}

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

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.events = eventRepo

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

	var updateCalls atomic.Int32
	mockJobRepo.On("UpdateFields", mock.Anything, uint64(8021), mock.Anything).Run(func(args mock.Arguments) {
		updateCalls.Add(1)
	}).Return(nil).Maybe()
	mockAssetRepo.On("SaveAssetSnapshot", mock.Anything, mock.Anything).Return(nil).Times(2)

	sink := newChunkedSyncSink(svc, job, account, ctx, collector.ResumeCursor{}, 0, time.Date(2026, 4, 9, 12, 0, 0, 0, time.UTC))
	first := &collector.AssetChunk{
		Result: collector.AssetResult{
			Assets: []caasmmodel.Asset{{NativeID: "asset-1", Name: "first"}},
		},
		Scope:    "region_service",
		Target:   "eu-west-1/ec2",
		Sequence: 1,
	}
	second := &collector.AssetChunk{
		Result: collector.AssetResult{
			Assets: []caasmmodel.Asset{{NativeID: "asset-2", Name: "second"}},
		},
		Scope:    "region_service",
		Target:   "eu-west-1/rds",
		Sequence: 2,
	}

	assert.NoError(t, sink.PersistChunk(ctx, first))
	for i := 0; i < 100; i++ {
		if updateCalls.Load() >= 2 {
			break
		}
		time.Sleep(10 * time.Millisecond)
	}
	if updateCalls.Load() < 2 {
		t.Fatalf("timed out waiting for first progress flush, updateCalls=%d", updateCalls.Load())
	}
	sink.mu.Lock()
	sink.lastProgressFlushAt = time.Now().Add(-31 * time.Second)
	sink.mu.Unlock()
	assert.NoError(t, sink.PersistChunk(ctx, second))
	assert.NoError(t, sink.Wait())

	assert.Equal(t, int32(4), updateCalls.Load())
	assert.Equal(t, 2, sink.chunkCount)
	assert.Equal(t, 2, job.Metadata["chunk_persist_count"])
	assert.Equal(t, 2, job.Metadata["last_chunk_sequence"])
	assert.Len(t, eventRepo.events, 2)
	assert.Equal(t, "Chunk sync prune completed", eventRepo.events[0].Message)
	assert.Equal(t, "Chunk persist finished", eventRepo.events[1].Message)
	mockAssetRepo.AssertExpectations(t)
	mockJobRepo.AssertExpectations(t)
}

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

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)

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

	mockJobRepo.On("UpdateFields", mock.Anything, uint64(803), mock.Anything).Return(nil).Maybe()

	sink := newChunkedSyncSink(svc, job, account, ctx, collector.ResumeCursor{}, 0, time.Date(2026, 4, 9, 12, 0, 0, 0, time.UTC))
	first := &collector.AssetChunk{
		Result:   collector.AssetResult{},
		Scope:    "route53_window",
		Target:   "example.com/window-1",
		Sequence: 1,
		Checkpoint: collector.ChunkCheckpoint{
			Route53ZoneID:               "Z1",
			Route53Zone:                 "example.com",
			Route53NextRecordName:       "www.example.com.",
			Route53NextRecordType:       "A",
			Route53NextRecordIdentifier: "weighted-1",
			Eligible:                    true,
		},
	}
	second := &collector.AssetChunk{
		Result:   collector.AssetResult{},
		Scope:    "route53_window",
		Target:   "example.com/window-2",
		Sequence: 2,
		Checkpoint: collector.ChunkCheckpoint{
			Route53ZoneID:       "Z1",
			Route53Zone:         "example.com",
			Route53ZoneComplete: true,
			Eligible:            true,
		},
	}

	assert.NoError(t, sink.persistChunk(ctx, first))
	firstCursor := resumeCursorFromMetadata(job.Metadata)
	progress, ok := firstCursor.Route53ZoneProgress["Z1"]
	if !ok {
		t.Fatalf("route53 progress missing after first chunk: %+v", firstCursor.Route53ZoneProgress)
	}
	assert.Equal(t, "example.com", progress.ZoneName)
	assert.Equal(t, "www.example.com.", progress.NextRecordName)
	assert.Equal(t, "A", progress.NextRecordType)
	assert.Equal(t, "weighted-1", progress.NextRecordIdentifier)
	assert.Empty(t, firstCursor.Route53CompletedZones)

	assert.NoError(t, sink.persistChunk(ctx, second))
	secondCursor := resumeCursorFromMetadata(job.Metadata)
	assert.Equal(t, []string{"Z1"}, secondCursor.Route53CompletedZones)
	if _, ok := secondCursor.Route53ZoneProgress["Z1"]; ok {
		t.Fatalf("route53 progress still present after zone completion: %+v", secondCursor.Route53ZoneProgress)
	}

	mockJobRepo.AssertExpectations(t)
}

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

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	job := &caasmmodel.SyncJob{
		Base:     coremodel.Base{ID: 804},
		Metadata: coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 102},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}

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

	var saveCalls atomic.Int32
	saveStarted := make(chan struct{})
	releaseFirst := make(chan struct{})
	mockAssetRepo.On("SaveAssetSnapshot", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
		if saveCalls.Add(1) == 1 {
			close(saveStarted)
			<-releaseFirst
		}
	}).Return(nil).Times(chunkPersistQueueCapacity + 2)

	sink := newChunkedSyncSink(svc, job, account, ctx, collector.ResumeCursor{}, 0, time.Date(2026, 4, 9, 12, 0, 0, 0, time.UTC))
	makeChunk := func(sequence int) *collector.AssetChunk {
		return &collector.AssetChunk{
			Result: collector.AssetResult{
				Assets: []caasmmodel.Asset{{
					NativeID: fmt.Sprintf("asset-%d", sequence),
					Name:     fmt.Sprintf("asset-%d", sequence),
				}},
			},
			Scope:    "region_service",
			Target:   fmt.Sprintf("eu-west-1/service-%d", sequence),
			Sequence: sequence,
		}
	}

	assert.NoError(t, sink.PersistChunk(ctx, makeChunk(1)))
	select {
	case <-saveStarted:
	case <-time.After(time.Second):
		t.Fatal("persist writer did not start first save")
	}

	for sequence := 2; sequence <= chunkPersistQueueCapacity+1; sequence++ {
		assert.NoError(t, sink.PersistChunk(ctx, makeChunk(sequence)))
	}

	extraCtx, cancelExtra := context.WithCancel(context.Background())
	extraDone := make(chan error, 1)
	go func() {
		extraDone <- sink.PersistChunk(extraCtx, makeChunk(chunkPersistQueueCapacity+2))
	}()

	select {
	case err := <-extraDone:
		t.Fatalf("expected extra PersistChunk to block under backpressure, got %v", err)
	case <-time.After(150 * time.Millisecond):
	}
	cancelExtra()

	select {
	case err := <-extraDone:
		t.Fatalf("expected extra PersistChunk to keep waiting on durable context, got %v", err)
	case <-time.After(150 * time.Millisecond):
	}

	close(releaseFirst)

	select {
	case err := <-extraDone:
		assert.NoError(t, err)
	case <-time.After(2 * time.Second):
		t.Fatal("extra PersistChunk did not resume after writer drain")
	}

	assert.NoError(t, sink.Wait())
	assert.Equal(t, chunkPersistQueueCapacity+2, sink.chunkCount)
	mockAssetRepo.AssertExpectations(t)
}

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

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.events = &stubSyncJobEventRepo{}

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

	mockJobRepo.On("UpdateFields", mock.Anything, uint64(806), mock.Anything).Return(nil).Maybe()

	saveStarted := make(chan struct{})
	releaseFirst := make(chan struct{})
	var saveCalls atomic.Int32
	mockAssetRepo.On("FindMinimalByNativeIDs", mock.Anything, uint64(104), []string{"asset-1"}).Return([]repository.AssetMinimal{}, nil).Once()
	mockAssetRepo.On("FindMinimalByNativeIDs", mock.Anything, uint64(104), []string{"asset-2"}).Return([]repository.AssetMinimal{}, nil).Once()
	mockAssetRepo.On("SaveAssetSnapshot", mock.Anything, mock.MatchedBy(func(asset *caasmmodel.Asset) bool {
		return asset != nil && asset.NativeID == "asset-1"
	})).Run(func(args mock.Arguments) {
		if saveCalls.Add(1) == 1 {
			close(saveStarted)
			<-releaseFirst
		}
	}).Return(nil).Once()
	mockAssetRepo.On("SaveAssetSnapshot", mock.Anything, mock.MatchedBy(func(asset *caasmmodel.Asset) bool {
		return asset != nil && asset.NativeID == "asset-2"
	})).Run(func(args mock.Arguments) {
		saveCalls.Add(1)
	}).Return(nil).Once()

	sink := newChunkedSyncSink(svc, job, account, ctx, collector.ResumeCursor{}, 0, time.Date(2026, 4, 9, 12, 0, 0, 0, time.UTC))
	first := &collector.AssetChunk{
		Result: collector.AssetResult{
			Assets: []caasmmodel.Asset{{NativeID: "asset-1", Name: "first"}},
		},
		Scope:    "region_service",
		Target:   "eu-west-1/ec2",
		Sequence: 1,
	}
	secondCtx, cancelSecond := context.WithCancel(context.Background())
	second := &collector.AssetChunk{
		Result: collector.AssetResult{
			Assets: []caasmmodel.Asset{{NativeID: "asset-2", Name: "second"}},
		},
		Scope:    "region_service",
		Target:   "eu-west-1/rds",
		Sequence: 2,
	}

	assert.NoError(t, sink.PersistChunk(ctx, first))
	select {
	case <-saveStarted:
	case <-time.After(time.Second):
		t.Fatal("persist writer did not start first save")
	}

	assert.NoError(t, sink.PersistChunk(secondCtx, second))
	cancelSecond()
	close(releaseFirst)

	assert.NoError(t, sink.Wait())
	assert.Equal(t, int32(2), saveCalls.Load())
	mockAssetRepo.AssertExpectations(t)
	mockJobRepo.AssertExpectations(t)
}

func TestChunkedSyncSinkStopsPersistWhenExecutionDeadlineExpires(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond)
	defer cancel()

	mockAcctRepo := new(mocks.MockCloudAccountRepository)
	mockJobRepo := new(mocks.MockSyncJobRepository)
	mockAssetRepo := new(mocks.MockAssetRepository)
	mockJobRepo.On("FindByID", mock.Anything, mock.AnythingOfType("uint64")).Return(&caasmmodel.SyncJob{}, nil).Maybe()

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)

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

	mockJobRepo.On("UpdateFields", mock.Anything, uint64(807), mock.Anything).Return(nil).Maybe()
	mockAssetRepo.On("FindMinimalByNativeIDs", mock.Anything, uint64(105), []string{"asset-1"}).Return([]repository.AssetMinimal{}, nil).Once()
	mockAssetRepo.On("SaveAssetSnapshot", mock.Anything, mock.MatchedBy(func(asset *caasmmodel.Asset) bool {
		return asset != nil && asset.NativeID == "asset-1"
	})).Run(func(args mock.Arguments) {
		persistCtx, ok := args.Get(0).(context.Context)
		if !ok {
			t.Fatalf("SaveAssetSnapshot ctx type = %T, want context.Context", args.Get(0))
		}
		<-persistCtx.Done()
	}).Return(context.DeadlineExceeded).Once()

	sink := newChunkedSyncSink(svc, job, account, ctx, collector.ResumeCursor{}, 0, time.Date(2026, 4, 9, 12, 0, 0, 0, time.UTC))
	chunk := &collector.AssetChunk{
		Result: collector.AssetResult{
			Assets: []caasmmodel.Asset{{NativeID: "asset-1", Name: "first"}},
		},
		Scope:    "region_service",
		Target:   "eu-west-1/ec2",
		Sequence: 1,
	}

	assert.NoError(t, sink.PersistChunk(ctx, chunk))
	assert.ErrorIs(t, sink.Wait(), context.DeadlineExceeded)
	mockAssetRepo.AssertExpectations(t)
	mockJobRepo.AssertExpectations(t)
}

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

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.SetConfigService(NewConfigService(&stubConfigRepoInternal{cfg: &caasmmodel.Config{Data: coremodel.JSONB{}}}))

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

	mockJobRepo.On("FindLatestResumableJobForAccount", mock.Anything, uint64(103), uint64(805), "").Return((*caasmmodel.SyncJob)(nil), nil).Once()
	mockJobRepo.On("UpdateFields", mock.Anything, uint64(805), mock.Anything).Return(nil)
	mockAcctRepo.On("Update", mock.Anything, mock.Anything).Return(nil).Once()
	mockAssetRepo.On("SoftDeleteStale", mock.Anything, uint64(103), mock.AnythingOfType("time.Time")).Return(nil).Once()

	saveStarted := make(chan struct{})
	releaseFirst := make(chan struct{})
	var saveCalls atomic.Int32
	mockAssetRepo.On("SaveAssetSnapshot", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
		if saveCalls.Add(1) == 1 {
			close(saveStarted)
			<-releaseFirst
		}
	}).Return(nil).Twice()

	svc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return &chunkAwareStubCollector{
			chunks: []*collector.AssetChunk{
				{
					Result: collector.AssetResult{
						Assets: []caasmmodel.Asset{{NativeID: "asset-1", Name: "first"}},
					},
					Scope:    "region_service",
					Target:   "eu-west-1/ec2",
					Sequence: 1,
				},
				{
					Result: collector.AssetResult{
						Assets: []caasmmodel.Asset{{NativeID: "asset-2", Name: "second"}},
					},
					Scope:    "region_service",
					Target:   "eu-west-1/rds",
					Sequence: 2,
					Checkpoint: collector.ChunkCheckpoint{
						Region:   "eu-west-1",
						Eligible: true,
					},
				},
			},
		}, "eu-west-1", nil
	}

	done := make(chan struct{})
	go func() {
		svc.runSync(ctx, job, account)
		close(done)
	}()

	select {
	case <-saveStarted:
	case <-time.After(time.Second):
		t.Fatal("runSync did not begin persisting the first chunk")
	}

	select {
	case <-done:
		t.Fatal("runSync returned before buffered chunk persistence drained")
	case <-time.After(150 * time.Millisecond):
	}

	close(releaseFirst)

	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("runSync did not finish after persistence drain completed")
	}

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

func TestChunkedSyncSyntheticCapacityRecovery(t *testing.T) {
	ctx := context.Background()
	totalChunks := 40
	persistedBeforeInterrupt := 15
	chunks := make([]*collector.AssetChunk, 0, totalChunks)
	for i := 0; i < totalChunks; i++ {
		scope := "region"
		target := fmt.Sprintf("eu-west-%d", (i%3)+1)
		checkpoint := collector.ChunkCheckpoint{
			Region:   target,
			Eligible: true,
		}
		if i%2 == 0 {
			scope = "route53"
			target = fmt.Sprintf("zone-%02d.example.com", i)
			checkpoint = collector.ChunkCheckpoint{
				Route53ZoneID:       fmt.Sprintf("Z%03d", i),
				Route53Zone:         target,
				Route53ZoneComplete: true,
				Eligible:            true,
			}
		}
		chunks = append(chunks, &collector.AssetChunk{
			Result: collector.AssetResult{
				Assets: []caasmmodel.Asset{{
					NativeID: fmt.Sprintf("asset-%02d", i),
					Name:     fmt.Sprintf("asset-%02d", i),
				}},
			},
			Scope:      scope,
			Target:     target,
			Sequence:   i + 1,
			Checkpoint: checkpoint,
		})
	}

	firstJobRepo := new(mocks.MockSyncJobRepository)
	firstAcctRepo := new(mocks.MockCloudAccountRepository)
	firstAssetRepo := new(mocks.MockAssetRepository)
	firstJobRepo.On("FindByID", mock.Anything, mock.AnythingOfType("uint64")).Return(&caasmmodel.SyncJob{}, nil).Maybe()
	firstSvc := NewCloudAccountService(firstAcctRepo, firstJobRepo, firstAssetRepo)
	firstSvc.SetConfigService(NewConfigService(&sequencedConfigRepo{
		cfgs: append([]*caasmmodel.Config{
			{Data: coremodel.JSONB{}},
			{Data: coremodel.JSONB{}},
			{Data: coremodel.JSONB{}},
		}, interruptConfigs(totalChunks, persistedBeforeInterrupt)...),
	}))
	firstCollector := &chunkAwareStubCollector{chunks: chunks}
	firstSvc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return firstCollector, "eu-west-1", nil
	}

	firstJob := &caasmmodel.SyncJob{Base: coremodel.Base{ID: 900}, Metadata: coremodel.JSONB{}}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 77},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}

	firstJobRepo.On("FindLatestResumableJobForAccount", mock.Anything, uint64(77), uint64(900), "").Return((*caasmmodel.SyncJob)(nil), nil).Once()
	firstJobRepo.On("FindByID", mock.Anything, uint64(900)).Return(&caasmmodel.SyncJob{}, nil).Maybe()
	firstJobRepo.On("UpdateFields", mock.Anything, uint64(900), mock.Anything).Return(nil)
	firstAcctRepo.On("Update", mock.Anything, mock.Anything).Return(nil).Once()
	firstAssetRepo.On("SaveAssetSnapshot", mock.Anything, mock.Anything).Return(nil)

	firstStart := time.Now()
	firstSvc.runSync(ctx, firstJob, account)
	firstDuration := time.Since(firstStart)

	firstCursor := resumeCursorFromMetadata(firstJob.Metadata)
	persistedChunkCount, ok := firstJob.Metadata["chunk_persist_count"].(int)
	if !ok {
		t.Fatalf("first run chunk_persist_count type = %T, want int", firstJob.Metadata["chunk_persist_count"])
	}
	if persistedChunkCount <= 0 || persistedChunkCount >= totalChunks {
		t.Fatalf("first run chunk_persist_count = %d, want between 1 and %d", persistedChunkCount, totalChunks-1)
	}
	if firstJob.Status != caasmmodel.SyncJobFailed {
		t.Fatalf("first run status = %s, want failed", firstJob.Status)
	}
	if firstJob.Metadata["stale_mark_completed"] == true {
		t.Fatalf("first run unexpectedly stale-marked")
	}
	if firstJob.Metadata["stale_mark_skipped_reason"] != "sync_error" {
		t.Fatalf("first run stale_mark_skipped_reason = %v, want sync_error", firstJob.Metadata["stale_mark_skipped_reason"])
	}
	t.Logf("partial synthetic capacity run: chunks=%d persisted=%d max_heap_alloc=%v duration_ms=%d", totalChunks, persistedChunkCount, firstJob.Metadata["chunk_max_heap_alloc"], firstDuration.Milliseconds())

	secondJobRepo := new(mocks.MockSyncJobRepository)
	secondAcctRepo := new(mocks.MockCloudAccountRepository)
	secondAssetRepo := new(mocks.MockAssetRepository)
	secondJobRepo.On("FindByID", mock.Anything, mock.AnythingOfType("uint64")).Return(&caasmmodel.SyncJob{}, nil).Maybe()
	secondSvc := NewCloudAccountService(secondAcctRepo, secondJobRepo, secondAssetRepo)
	secondSvc.SetConfigService(NewConfigService(&stubConfigRepoInternal{cfg: &caasmmodel.Config{Data: coremodel.JSONB{}}}))
	secondCollector := &chunkAwareStubCollector{chunks: chunks[persistedChunkCount:]}
	secondSvc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return secondCollector, "eu-west-1", nil
	}

	secondJob := &caasmmodel.SyncJob{Base: coremodel.Base{ID: 901}, Metadata: coremodel.JSONB{}}
	priorJob := caasmmodel.SyncJob{
		Base:     coremodel.Base{ID: 900},
		Status:   caasmmodel.SyncJobPartial,
		Metadata: firstJob.Metadata,
	}
	resumeWindow := resumeWindowStart(firstCursor, time.Time{})

	secondJobRepo.On("FindLatestResumableJobForAccount", mock.Anything, uint64(77), uint64(901), "").Return(&priorJob, nil).Once()
	secondJobRepo.On("FindByID", mock.Anything, uint64(901)).Return(&caasmmodel.SyncJob{}, nil).Maybe()
	secondJobRepo.On("UpdateFields", mock.Anything, uint64(901), mock.Anything).Return(nil)
	secondAcctRepo.On("Update", mock.Anything, mock.Anything).Return(nil).Once()
	secondAssetRepo.On("SaveAssetSnapshot", mock.Anything, mock.Anything).Return(nil)
	secondAssetRepo.On("SoftDeleteStale", mock.Anything, uint64(77), mock.MatchedBy(func(cutoff time.Time) bool {
		return cutoff.Equal(resumeWindow)
	})).Return(nil).Once()

	secondStart := time.Now()
	secondSvc.runSync(ctx, secondJob, account)
	secondDuration := time.Since(secondStart)

	secondCursor := resumeCursorFromMetadata(secondJob.Metadata)
	if secondJob.Status != caasmmodel.SyncJobSuccess {
		t.Fatalf("second run status = %s, want success", secondJob.Status)
	}
	if secondJob.Metadata["stale_mark_completed"] != true {
		t.Fatalf("second run did not stale-mark after recovery")
	}
	if secondCollector.resume == nil {
		t.Fatalf("second collector did not receive resume cursor")
	}
	if secondCollector.resume.WindowStartedAt != firstCursor.WindowStartedAt {
		t.Fatalf("resume window = %s, want %s", secondCollector.resume.WindowStartedAt, firstCursor.WindowStartedAt)
	}
	if len(secondCursor.CompletedRegions) < len(firstCursor.CompletedRegions) {
		t.Fatalf("second cursor completed regions = %v, want at least prior %v", secondCursor.CompletedRegions, firstCursor.CompletedRegions)
	}
	if len(secondCursor.Route53CompletedZones) < len(firstCursor.Route53CompletedZones) {
		t.Fatalf("second cursor route53 zones = %v, want at least prior %v", secondCursor.Route53CompletedZones, firstCursor.Route53CompletedZones)
	}
	t.Logf("recovery synthetic capacity run: chunks=%v max_heap_alloc=%v duration_ms=%d stale_mark=%v", secondJob.Metadata["chunk_persist_count"], secondJob.Metadata["chunk_max_heap_alloc"], secondDuration.Milliseconds(), secondJob.Metadata["stale_mark_completed"])
}

func interruptConfigs(totalChunks, persistedBeforeInterrupt int) []*caasmmodel.Config {
	configs := make([]*caasmmodel.Config, 0, totalChunks)
	for i := 0; i < totalChunks; i++ {
		data := coremodel.JSONB{}
		if i >= persistedBeforeInterrupt {
			data["sync_settings"] = map[string]interface{}{"kill_switch_enabled": true}
		}
		configs = append(configs, &caasmmodel.Config{Data: data})
	}
	return configs
}

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

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.SetConfigService(NewConfigService(&stubConfigRepoInternal{
		cfg: &caasmmodel.Config{
			Data: coremodel.JSONB{
				"collection_scope": map[string]interface{}{
					"ip_collection_scope":   string(caasmmodel.CollectionScopePublicOnly),
					"port_collection_scope": string(caasmmodel.CollectionScopePublicOnly),
				},
			},
		},
	}))

	chunkCollector := &chunkAwareStubCollector{
		chunks: []*collector.AssetChunk{
			{
				Result:   collector.AssetResult{},
				Scope:    "route53",
				Target:   "example.com",
				Sequence: 1,
				Checkpoint: collector.ChunkCheckpoint{
					Route53ZoneID:       "Z1",
					Route53Zone:         "example.com",
					Route53ZoneComplete: true,
					Eligible:            true,
				},
			},
			{
				Result:   collector.AssetResult{},
				Scope:    "route53",
				Target:   "api.example.com",
				Sequence: 2,
				Checkpoint: collector.ChunkCheckpoint{
					Route53ZoneID:       "Z2",
					Route53Zone:         "api.example.com",
					Route53ZoneComplete: true,
					Eligible:            true,
				},
			},
		},
		partialFailures: []collector.CollectionIssue{
			{Scope: "route53", Target: "api.example.com", Message: "simulated collection failure"},
		},
	}
	svc.collectorBuilder = func(ctx context.Context, account *caasmmodel.CloudAccount) (collector.Collector, string, error) {
		return chunkCollector, "us-east-1", nil
	}

	job := &caasmmodel.SyncJob{
		Base: coremodel.Base{ID: 802},
		Metadata: coremodel.JSONB{
			"resume_cursor": resumeCursorJSONB(collector.ResumeCursor{
				WindowStartedAt: "2026-04-09T12:00:00Z",
			}),
		},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 8},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}

	mockJobRepo.On("UpdateFields", mock.Anything, uint64(802), mock.Anything).Return(nil)
	mockAcctRepo.On("Update", mock.Anything, mock.Anything).Return(nil).Once()
	mockAssetRepo.On("DeletePrivateIPAssetsForCloudAccount", mock.Anything, uint64(8)).Return(nil).Once()

	svc.runSync(ctx, job, account)

	assert.Equal(t, caasmmodel.SyncJobPartial, job.Status)
	assert.Equal(t, false, job.Metadata["stale_mark_eligible"])
	assert.Equal(t, false, job.Metadata["full_run_completed"])
	assert.Equal(t, false, job.Metadata["stale_mark_completed"])
	assert.Equal(t, "partial_failures", job.Metadata["stale_mark_skipped_reason"])
	assert.NotNil(t, chunkCollector.resume)
	assert.Equal(t, "2026-04-09T12:00:00Z", chunkCollector.resume.WindowStartedAt)
	assert.Equal(t, collector.ResumeCursor{
		WindowStartedAt:       "2026-04-09T12:00:00Z",
		Route53CompletedZones: []string{"Z1", "Z2"},
	}, resumeCursorFromMetadata(job.Metadata))
	mockAssetRepo.AssertExpectations(t)
	mockJobRepo.AssertExpectations(t)
	mockAcctRepo.AssertExpectations(t)
}

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

	svc := NewCloudAccountService(mockAcctRepo, mockJobRepo, mockAssetRepo)
	svc.SetConfigService(NewConfigService(&stubConfigRepoInternal{
		cfg: &caasmmodel.Config{
			Data: coremodel.JSONB{
				"collection_scope": map[string]interface{}{
					"ip_collection_scope":   string(caasmmodel.CollectionScopePublicOnly),
					"port_collection_scope": string(caasmmodel.CollectionScopePublicOnly),
				},
			},
		},
	}))

	resumeCursor := collector.ResumeCursor{
		WindowStartedAt:       "2026-04-09T12:00:00Z",
		CompletedGlobal:       []string{"route53"},
		CompletedRegions:      []string{"eu-west-2"},
		Route53CompletedZones: []string{"Z1"},
	}
	previousJob := caasmmodel.SyncJob{
		Base:   coremodel.Base{ID: 601},
		Status: caasmmodel.SyncJobPartial,
		Metadata: coremodel.JSONB{
			"resume_cursor": resumeCursorJSONB(resumeCursor),
		},
	}

	chunkCollector := &chunkAwareStubCollector{
		chunks: []*collector.AssetChunk{
			{
				Result:   collector.AssetResult{},
				Scope:    "route53",
				Target:   "api.example.com",
				Sequence: 2,
				Checkpoint: collector.ChunkCheckpoint{
					Route53ZoneID:       "Z2",
					Route53Zone:         "api.example.com",
					Route53ZoneComplete: true,
					Eligible:            true,
				},
			},
		},
	}
	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: 803},
		Metadata: coremodel.JSONB{},
	}
	account := &caasmmodel.CloudAccount{
		Base:     coremodel.Base{ID: 8},
		Provider: caasmmodel.CloudProviderAWS,
		Metadata: coremodel.JSONB{},
	}

	mockJobRepo.On("FindLatestResumableJobForAccount", ctx, uint64(8), uint64(803), "").Return(&previousJob, nil)
	mockJobRepo.On("UpdateFields", mock.Anything, uint64(803), mock.Anything).Return(nil)
	mockAcctRepo.On("Update", mock.Anything, mock.Anything).Return(nil).Once()
	mockAssetRepo.On("DeletePrivateIPAssetsForCloudAccount", mock.Anything, uint64(8)).Return(nil).Once()
	mockAssetRepo.On("SoftDeleteStale", ctx, uint64(8), mock.Anything).Return(nil).Once()

	svc.runSync(ctx, job, account)

	assert.NotNil(t, chunkCollector.resume)
	assert.Equal(t, resumeCursor, *chunkCollector.resume)
	assert.Equal(t, uint64(601), job.Metadata["resume_source_job_id"])
	assert.Equal(t, true, job.Metadata["stale_mark_completed"])
	assert.Equal(t, true, job.Metadata["full_run_completed"])
	assert.Equal(t, true, job.Metadata["stale_mark_eligible"])
	assert.Equal(t, caasmmodel.SyncJobSuccess, job.Status)
	assert.Equal(t, collector.ResumeCursor{
		WindowStartedAt:       "2026-04-09T12:00:00Z",
		CompletedGlobal:       []string{"route53"},
		CompletedRegions:      []string{"eu-west-2"},
		Route53CompletedZones: []string{"Z1", "Z2"},
	}, resumeCursorFromMetadata(job.Metadata))
	mockJobRepo.AssertExpectations(t)
	mockAcctRepo.AssertExpectations(t)
	mockAssetRepo.AssertExpectations(t)
}