package records

import (
	"context"
	"crypto/sha256"
	"encoding/csv"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"maps"
	"strings"
	"time"

	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/worker"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
)

const (
	RuntimeBulkIngestWorkFamily     = "runtime.bulk-ingest.v1"
	DefaultBulkIngestCheckpointRows = 100
)

// BulkExecutionBinder reopens an immutable mutation view. permanent is
// meaningful only when err is non-nil; temporary artifact access must remain
// retryable while a scope/digest contradiction must fail the job.
type BulkExecutionBinder func(context.Context, catalogsnapshot.RuntimeWorkBinding) (bound context.Context, permanent bool, err error)

var (
	ErrBulkObjectNotFound   = errors.New("bulk object not found")
	errBulkCSVOffsetPastEnd = errors.New("bulk CSV durable row offset is past end of object")
)

type BulkObjectReader interface {
	OpenBulkObject(context.Context, string) (io.ReadCloser, int64, error)
}

type BulkStorageAuthority struct {
	Provider string
	Bucket   string
	Digest   string
}

type BulkMutationExecutor interface {
	LoadUserSession(context.Context, string) (*models.UserSession, error)
	RecordMutationTransactionObjectNames(context.Context, string, semantic.OperationKind) []string
	RunRecordMutationTransactionWithOutcome(context.Context, []string, func(context.Context) error) (ports.TransactionCompletionOutcome, error)
	BulkInsert(context.Context, string, []models.Record, *models.UserSession, BulkInsertOptions) (BulkInsertResult, error)
	Update(context.Context, string, string, models.Record, *models.UserSession) error
	Delete(context.Context, string, string, *models.UserSession) error
	UpsertByExternalID(context.Context, string, string, string, models.Record, *models.UserSession) (models.Record, bool, error)
}

type BulkWorkSource struct {
	store       *persistence.BulkWorkStore
	records     BulkMutationExecutor
	objects     BulkObjectReader
	storage     BulkStorageAuthority
	authority   string
	binder      BulkExecutionBinder
	chunkSize   int
	maxAttempts int
}

func NewBulkWorkSource(
	store *persistence.BulkWorkStore,
	records BulkMutationExecutor,
	objects BulkObjectReader,
	storageAuthority BulkStorageAuthority,
	binder BulkExecutionBinder,
	chunkSize, maxAttempts int,
) (*BulkWorkSource, error) {
	if store == nil || records == nil || objects == nil || binder == nil || chunkSize <= 0 || maxAttempts <= 0 {
		return nil, errors.New("bulk work source: durable store, mutation graph, object store, binder, and bounds are required")
	}
	if strings.TrimSpace(storageAuthority.Provider) == "" ||
		strings.TrimSpace(storageAuthority.Bucket) == "" ||
		strings.TrimSpace(storageAuthority.Digest) == "" {
		return nil, errors.New("bulk work source: exact storage authority is required")
	}
	return &BulkWorkSource{
		store: store, records: records, objects: objects, storage: storageAuthority,
		authority: storageAuthority.Digest, binder: binder, chunkSize: chunkSize, maxAttempts: maxAttempts,
	}, nil
}

func (s *BulkWorkSource) CheckReady(ctx context.Context) error {
	if err := s.store.CheckReady(ctx); err != nil {
		return err
	}
	return s.store.CheckStorageAuthority(ctx, s.storage.Provider, s.storage.Bucket, s.authority)
}

func (s *BulkWorkSource) Reconcile(ctx context.Context) error { return s.store.Reconcile(ctx) }

func (s *BulkWorkSource) Claim(ctx context.Context, capacity int) (worker.ClaimBatch, error) {
	claims, more, err := s.store.Claim(ctx, capacity)
	if err != nil {
		return worker.ClaimBatch{}, err
	}
	batch := worker.ClaimBatch{More: more, Attempts: make([]worker.Attempt, 0, len(claims))}
	for _, value := range claims {
		claim := value
		batch.Attempts = append(batch.Attempts, worker.Attempt{
			ID:  claim.Job.ID,
			Run: func(runCtx context.Context) error { return s.run(runCtx, claim) },
		})
	}
	return batch, nil
}

func (s *BulkWorkSource) NextWorkAt(ctx context.Context) (time.Time, bool, error) {
	return s.store.NextWorkAt(ctx)
}

func (s *BulkWorkSource) ReadGeneration(ctx context.Context, _ uint64) (uint64, error) {
	return s.store.ReadGeneration(ctx)
}

func (s *BulkWorkSource) run(ctx context.Context, claim persistence.BulkWorkClaim) error {
	binding, err := catalogsnapshot.OpenRuntimeWorkBinding(claim.Job.WorkBinding)
	if err != nil {
		return s.fail(ctx, claim, fmt.Errorf("invalid immutable metadata binding: %w", err))
	}
	boundCtx, permanent, err := s.binder(ctx, binding)
	if err != nil || boundCtx == nil {
		if err == nil {
			err = errors.New("metadata binder returned no context")
			permanent = true
		}
		cause := fmt.Errorf("reopen immutable metadata: %w", err)
		if permanent {
			return s.fail(ctx, claim, cause)
		}
		return s.retry(ctx, claim, cause)
	}
	user, err := s.records.LoadUserSession(boundCtx, strings.TrimSpace(claim.Job.CreatedById))
	if err != nil {
		return s.retry(boundCtx, claim, fmt.Errorf("re-establish current job owner authority: %w", err))
	}
	if user == nil || strings.TrimSpace(user.ID) != strings.TrimSpace(claim.Job.CreatedById) {
		return s.fail(boundCtx, claim, errors.New("re-establish current job owner authority: job owner is unavailable or inconsistent"))
	}
	if !claim.Job.Operation.IsIngest() {
		return s.fail(boundCtx, claim, fmt.Errorf("unsupported bulk operation %q", claim.Job.Operation))
	}
	batches, err := s.store.GetBatches(boundCtx, claim.Job.ID)
	if err != nil {
		return s.retry(boundCtx, claim, fmt.Errorf("load batches: %w", err))
	}
	processed, failed := 0, 0
	for _, batch := range batches {
		if batch.State == "Completed" {
			processed += batch.RecordsProcessed
			failed += batch.RecordsFailed
			continue
		}
		if err := s.validateObject(boundCtx, batch); err != nil {
			if isPermanentBulkWorkError(err) {
				return s.fail(boundCtx, claim, err)
			}
			return s.retry(boundCtx, claim, err)
		}
		if err := s.processBatch(boundCtx, claim, claim.Job, batch, user); err != nil {
			if errors.Is(err, persistence.ErrBulkIngestJobAborted) {
				return nil
			}
			// A checkpoint transaction error may have an outcome the caller cannot
			// safely infer. Returning it makes the controller fail the process;
			// lease expiry and fencing decide the next owner after restart.
			if isUnknownBulkOutcomeError(err) {
				return err
			}
			if isPermanentBulkWorkError(err) {
				return s.fail(boundCtx, claim, err)
			}
			return s.retry(boundCtx, claim, err)
		}
		processed += batch.RecordsProcessed
		failed += batch.RecordsFailed
	}
	err = s.store.Complete(boundCtx, claim, processed+failed, failed)
	if errors.Is(err, persistence.ErrBulkIngestJobAborted) {
		return nil
	}
	return err
}

func (s *BulkWorkSource) validateObject(ctx context.Context, batch *models.BulkBatch) error {
	if batch == nil || batch.ObjectKey == "" || batch.ContentSHA256 == "" ||
		batch.StorageProvider != s.storage.Provider || batch.StorageBucket != s.storage.Bucket ||
		batch.StorageAuthorityDigest != s.authority {
		return permanentBulkWorkError{errors.New("bulk batch storage authority is incomplete or does not match this deployment")}
	}
	body, objectSize, err := s.objects.OpenBulkObject(ctx, batch.ObjectKey)
	if err != nil {
		if errors.Is(err, ErrBulkObjectNotFound) {
			return permanentBulkWorkError{fmt.Errorf("bulk batch object is missing: %w", err)}
		}
		return fmt.Errorf("open bulk batch object: %w", err)
	}
	hash := sha256.New()
	written, copyErr := io.Copy(hash, body)
	closeErr := body.Close()
	if copyErr != nil {
		return fmt.Errorf("verify bulk batch object: %w", copyErr)
	}
	if closeErr != nil {
		return fmt.Errorf("close bulk batch object: %w", closeErr)
	}
	if objectSize != batch.ContentLength || written != batch.ContentLength ||
		hex.EncodeToString(hash.Sum(nil)) != batch.ContentSHA256 {
		return permanentBulkWorkError{errors.New("bulk batch object content does not match its durable checksum and length")}
	}
	return nil
}

func (s *BulkWorkSource) processBatch(
	ctx context.Context,
	claim persistence.BulkWorkClaim,
	job *models.BulkJob,
	batch *models.BulkBatch,
	user *models.UserSession,
) error {
	body, _, err := s.objects.OpenBulkObject(ctx, batch.ObjectKey)
	if err != nil {
		return fmt.Errorf("reopen bulk batch object: %w", err)
	}
	defer body.Close()
	reader := csv.NewReader(body)
	headers, err := reader.Read()
	if err != nil {
		var parseErr *csv.ParseError
		if errors.As(err, &parseErr) || (errors.Is(err, io.EOF) && batch.ContentLength == 0) {
			return permanentBulkWorkError{fmt.Errorf("read CSV header: %w", err)}
		}
		return fmt.Errorf("read CSV header: %w", err)
	}
	for index := range headers {
		headers[index] = strings.ToLower(strings.TrimSpace(headers[index]))
	}
	if err := skipBulkCSVRows(reader, batch.NextRowOffset); err != nil {
		cause := fmt.Errorf("resume CSV at durable row %d: %w", batch.NextRowOffset, err)
		if errors.Is(err, errBulkCSVOffsetPastEnd) {
			return permanentBulkWorkError{cause}
		}
		return cause
	}
	if err := s.store.RunClaimTransaction(ctx, claim, func(txCtx context.Context, _ *persistence.BulkRepository) error {
		return s.store.UpdateBatchCheckpoint(txCtx, batch.ID, batch.NextRowOffset,
			batch.RecordsProcessed, batch.RecordsFailed, batch.TotalRecords, "InProgress", batch.LastError)
	}); err != nil {
		if errors.Is(err, persistence.ErrBulkClaimTransactionRolledBack) {
			return err
		}
		return unknownBulkOutcomeError{err}
	}

	for {
		readLimit := s.chunkSize
		if job.Operation != models.OperationInsert {
			// Independent record mutations own one effect/checkpoint transaction
			// per CSV row. That is the established Runtime Record Mutation model:
			// a failed row rolls back completely before its terminal failure receipt
			// is persisted, while successful effects and cursors commit together.
			readLimit = 1
		}
		chunk, parseFailures, eof, err := readBulkCSVChunk(reader, headers, readLimit)
		if err != nil {
			return fmt.Errorf("read CSV rows: %w", err)
		}
		if len(chunk) == 0 && parseFailures == 0 && eof {
			if err := s.store.RunClaimTransaction(ctx, claim, func(txCtx context.Context, _ *persistence.BulkRepository) error {
				return s.store.UpdateBatchCheckpoint(txCtx, batch.ID, batch.NextRowOffset,
					batch.RecordsProcessed, batch.RecordsFailed, batch.TotalRecords, "Completed", batch.LastError)
			}); err != nil {
				if errors.Is(err, persistence.ErrBulkClaimTransactionRolledBack) {
					return err
				}
				return unknownBulkOutcomeError{err}
			}
			break
		}
		consumed := int64(len(chunk) + parseFailures)
		var committed bulkBatchCheckpoint
		transactionObjects := s.records.RecordMutationTransactionObjectNames(
			ctx, job.Object, bulkSemanticOperationKind(job.Operation),
		)
		completion, mutationErr := s.records.RunRecordMutationTransactionWithOutcome(
			ctx, transactionObjects, func(recordsCtx context.Context) error {
				return s.store.RunClaimInTransaction(recordsCtx, claim, func(txCtx context.Context, _ *persistence.BulkRepository) error {
					// The Records owner may replay this callback after learning parent
					// prelocks. Never let a rolled-back attempt's generated IDs or
					// canonicalized values become the next attempt's input.
					attemptRows := cloneBulkWorkRecords(chunk)
					successCount, failureCount, errorsForRows, applyErr := s.applyChunk(txCtx, job, attemptRows, user)
					if applyErr != nil {
						return applyErr
					}
					committed = nextBulkBatchCheckpoint(
						batch, consumed, successCount, failureCount+parseFailures, errorsForRows, eof,
					)
					return s.store.UpdateBatchCheckpoint(txCtx, batch.ID, committed.nextRowOffset,
						committed.recordsProcessed, committed.recordsFailed, committed.totalRecords,
						committed.state, committed.lastError)
				})
			},
		)
		err = mutationErr
		if err != nil {
			var domain bulkRowDomainFailure
			var chunk *bulkChunkDomainFailure
			checkpointFailures := 0
			var checkpointMessages []string
			if completion == ports.TransactionCompletionRolledBack && errors.As(err, &chunk) {
				// BulkInsert's aggregate error is the instruction that rolls
				// back transaction-local work such as autonumber allocation.
				checkpointFailures = chunk.failed
				checkpointMessages = chunk.messages
			} else if completion == ports.TransactionCompletionRolledBack && errors.As(err, &domain) {
				// The business effect is proven absent. Persist only the failed-row
				// cursor in a fresh claim transaction, matching Runtime Record
				// Mutation's durable independent-operation receipt.
				checkpointFailures = 1
				checkpointMessages = []string{domain.Error()}
			}
			if checkpointFailures > 0 {
				// Only after rollback is proven may rejected rows move the
				// durable cursor in this fresh claim transaction.
				committed = nextBulkBatchCheckpoint(
					batch, consumed, 0, checkpointFailures+parseFailures, checkpointMessages, eof,
				)
				if checkpointErr := s.store.RunClaimTransaction(ctx, claim, func(txCtx context.Context, _ *persistence.BulkRepository) error {
					return s.store.UpdateBatchCheckpoint(txCtx, batch.ID, committed.nextRowOffset,
						committed.recordsProcessed, committed.recordsFailed, committed.totalRecords,
						committed.state, committed.lastError)
				}); checkpointErr != nil {
					if errors.Is(checkpointErr, persistence.ErrBulkClaimTransactionRolledBack) {
						return checkpointErr
					}
					return unknownBulkOutcomeError{checkpointErr}
				}
				err = nil
			} else if completion == ports.TransactionCompletionRolledBack {
				return errors.Join(persistence.ErrBulkClaimTransactionRolledBack, err)
			} else {
				return unknownBulkOutcomeError{err}
			}
		}
		// TransactionManager can replay the callback on a lock retry. Only copy
		// the winning candidate into process memory after commit succeeds.
		batch.NextRowOffset = committed.nextRowOffset
		batch.RecordsProcessed = committed.recordsProcessed
		batch.RecordsFailed = committed.recordsFailed
		batch.TotalRecords = committed.totalRecords
		batch.LastError = committed.lastError
		if eof {
			break
		}
	}
	return nil
}

func skipBulkCSVRows(reader *csv.Reader, count int64) error {
	for skipped := int64(0); skipped < count; skipped++ {
		_, err := reader.Read()
		if errors.Is(err, io.EOF) {
			return errBulkCSVOffsetPastEnd
		}
		// A malformed CSV record is still one durable cursor unit. The initial
		// pass counted it as a failed row, so recovery must advance past it too.
		var parseErr *csv.ParseError
		if errors.As(err, &parseErr) {
			continue
		}
		if err != nil {
			return err
		}
	}
	return nil
}

type bulkBatchCheckpoint struct {
	nextRowOffset    int64
	recordsProcessed int
	recordsFailed    int
	totalRecords     int
	state            string
	lastError        string
}

func nextBulkBatchCheckpoint(
	base *models.BulkBatch,
	consumed int64,
	succeeded, failed int,
	rowErrors []string,
	eof bool,
) bulkBatchCheckpoint {
	state := "InProgress"
	if eof {
		state = "Completed"
	}
	return bulkBatchCheckpoint{
		nextRowOffset:    base.NextRowOffset + consumed,
		recordsProcessed: base.RecordsProcessed + succeeded,
		recordsFailed:    base.RecordsFailed + failed,
		totalRecords:     base.TotalRecords + int(consumed),
		state:            state,
		lastError:        strings.Join(rowErrors, "; "),
	}
}

func readBulkCSVChunk(reader *csv.Reader, headers []string, limit int) ([]models.Record, int, bool, error) {
	records := make([]models.Record, 0, limit)
	parseFailures := 0
	for len(records)+parseFailures < limit {
		row, err := reader.Read()
		if errors.Is(err, io.EOF) {
			return records, parseFailures, true, nil
		}
		var parseErr *csv.ParseError
		if errors.As(err, &parseErr) {
			parseFailures++
			continue
		}
		if err != nil {
			// No row in this chunk has been checkpointed yet. Return no partial
			// outcome so a reopened object can replay the whole chunk.
			return nil, 0, false, err
		}
		record := make(models.Record, len(row))
		for index, value := range row {
			if index < len(headers) && headers[index] != "" {
				record[headers[index]] = value
			}
		}
		records = append(records, record)
	}
	return records, parseFailures, false, nil
}

func (s *BulkWorkSource) applyChunk(
	ctx context.Context,
	job *models.BulkJob,
	rows []models.Record,
	user *models.UserSession,
) (int, int, []string, error) {
	if len(rows) == 0 {
		return 0, 0, nil, nil
	}
	switch job.Operation {
	case models.OperationInsert:
		result, err := s.records.BulkInsert(ctx, job.Object, rows, user, BulkInsertOptions{
			BatchSize: 100, NormalizeFromCSV: true, AllowSystemFields: true,
			AllowLegacyListViewBulkIngest: strings.EqualFold(strings.TrimSpace(job.Object), constants.TableListView),
		})
		return completeBulkInsertOutcome(job.Object, result, len(rows), err)
	case models.OperationUpdate:
		if len(rows) != 1 {
			return 0, 0, nil, fmt.Errorf("bulk independent update requires exactly one row, got %d", len(rows))
		}
		row := maps.Clone(rows[0])
		id := bulkCSVRecordID(row)
		if id == "" {
			return 0, 1, []string{"missing Id field"}, nil
		}
		delete(row, "id")
		delete(row, "Id")
		if err := s.records.Update(ctx, job.Object, id, row, user); err != nil {
			return 0, 0, nil, classifyBulkRowMutationError(job, id, err)
		}
		return 1, 0, nil, nil
	case models.OperationDelete:
		if len(rows) != 1 {
			return 0, 0, nil, fmt.Errorf("bulk independent delete requires exactly one row, got %d", len(rows))
		}
		id := bulkCSVRecordID(rows[0])
		if id == "" {
			return 0, 1, []string{"missing Id field"}, nil
		}
		if err := s.records.Delete(ctx, job.Object, id, user); err != nil {
			return 0, 0, nil, classifyBulkRowMutationError(job, id, err)
		}
		return 1, 0, nil, nil
	case models.OperationUpsert:
		if len(rows) != 1 {
			return 0, 0, nil, fmt.Errorf("bulk independent upsert requires exactly one row, got %d", len(rows))
		}
		externalID := strings.ToLower(strings.TrimSpace(job.ExternalIdFieldName))
		if externalID == "" {
			externalID = "id"
		}
		row := rows[0]
		value, ok := row[externalID]
		if !ok || strings.TrimSpace(fmt.Sprint(value)) == "" {
			return 0, 1, []string{fmt.Sprintf("missing external ID field %s", externalID)}, nil
		}
		if _, _, err := s.records.UpsertByExternalID(ctx, job.Object, externalID, fmt.Sprint(value), row, user); err != nil {
			return 0, 0, nil, classifyBulkRowMutationError(job, fmt.Sprint(value), err)
		}
		return 1, 0, nil, nil
	default:
		return 0, 0, nil, permanentBulkWorkError{fmt.Errorf("unsupported bulk operation %q", job.Operation)}
	}
}

// completeBulkInsertOutcome adapts BulkInsert's established result+error shape
// to the worker's stricter checkpoint contract. A top-level error always leaves
// this transaction: BulkInsert uses that error to request rollback of
// transaction-local work performed before final row validation. A complete
// all-row domain rejection carries its row outcomes through that rollback so a
// fresh transaction can checkpoint them after rollback is proven.
func completeBulkInsertOutcome(objectName string, result BulkInsertResult, requested int, mutationErr error) (int, int, []string, error) {
	complete := requested >= 0 && result.SuccessCount >= 0 && result.FailedCount >= 0 &&
		result.SuccessCount+result.FailedCount == requested &&
		len(result.Errors) == result.FailedCount
	if mutationErr != nil {
		// A complete all-row domain rejection is checkpointable only after the
		// Records owner proves this transaction rolled back. It is not evidence
		// that the failure happened before transactional side effects: final
		// validation runs after autonumber assignment.
		_, knownDomainRejection := classifyRuntimeMutationDomainError(semantic.Operation{
			Kind: semantic.OperationCreate, ObjectID: objectName,
		}, 0, mutationErr)
		allRowsRejected := complete && result.SuccessCount == 0 &&
			result.FailedCount == requested && knownDomainRejection
		if allRowsRejected {
			return 0, 0, nil, &bulkChunkDomainFailure{
				cause: mutationErr, failed: result.FailedCount,
				messages: append([]string(nil), result.Errors...),
			}
		}
		return 0, 0, nil, mutationErr
	}
	if !complete {
		return 0, 0, nil, fmt.Errorf(
			"bulk insert returned incomplete row outcomes: requested=%d success=%d failed=%d errors=%d",
			requested, result.SuccessCount, result.FailedCount, len(result.Errors),
		)
	}
	return result.SuccessCount, result.FailedCount, append([]string(nil), result.Errors...), nil
}

func bulkCSVRecordID(row models.Record) string {
	for _, key := range []string{"id", "Id"} {
		if value, ok := row[key]; ok {
			return strings.TrimSpace(fmt.Sprint(value))
		}
	}
	return ""
}

func cloneBulkWorkRecords(records []models.Record) []models.Record {
	cloned := make([]models.Record, len(records))
	for index, record := range records {
		cloned[index] = maps.Clone(record)
	}
	return cloned
}

type bulkRowDomainFailure struct{ error }

type bulkChunkDomainFailure struct {
	cause    error
	failed   int
	messages []string
}

func (e *bulkChunkDomainFailure) Error() string { return e.cause.Error() }
func (e *bulkChunkDomainFailure) Unwrap() error { return e.cause }

func bulkSemanticOperationKind(operation models.BulkOperation) semantic.OperationKind {
	switch operation {
	case models.OperationInsert:
		return semantic.OperationCreate
	case models.OperationDelete:
		return semantic.OperationDelete
	default:
		// Update and either branch of Upsert mutate the declared object. The
		// transaction object inventory differs only for metadata cascades on
		// delete, so Update is the neutral non-delete classification here.
		return semantic.OperationUpdate
	}
}

func classifyBulkRowMutationError(job *models.BulkJob, identity string, err error) error {
	if err == nil {
		return nil
	}
	kinds := []semantic.OperationKind{semantic.OperationUpdate}
	if job != nil {
		switch job.Operation {
		case models.OperationDelete:
			kinds = []semantic.OperationKind{semantic.OperationDelete}
		case models.OperationUpsert:
			// Upsert can fail on either its create probe or adopted update path.
			kinds = []semantic.OperationKind{semantic.OperationUpdate, semantic.OperationCreate}
		}
	}
	for _, kind := range kinds {
		operation := semantic.Operation{Kind: kind, RecordID: identity}
		if job != nil {
			operation.ObjectID = job.Object
		}
		if _, domain := classifyRuntimeMutationDomainError(operation, 0, err); domain {
			return bulkRowDomainFailure{err}
		}
	}
	return err
}

func (s *BulkWorkSource) retry(ctx context.Context, claim persistence.BulkWorkClaim, cause error) error {
	if claim.Job.RetryCount+1 >= s.maxAttempts {
		return s.fail(ctx, claim, cause)
	}
	delay := time.Second << min(claim.Job.RetryCount, 6)
	return consumeBulkAbort(s.store.Retry(ctx, claim, cause, delay))
}

func (s *BulkWorkSource) fail(ctx context.Context, claim persistence.BulkWorkClaim, cause error) error {
	return consumeBulkAbort(s.store.Fail(ctx, claim, cause))
}

func consumeBulkAbort(err error) error {
	if errors.Is(err, persistence.ErrBulkIngestJobAborted) {
		return nil
	}
	return err
}

type permanentBulkWorkError struct{ error }

func isPermanentBulkWorkError(err error) bool {
	var target permanentBulkWorkError
	return errors.As(err, &target)
}

// unknownBulkOutcomeError prevents a second durable transition after a
// checkpoint transaction returned an outcome the worker cannot prove. The
// shared controller treats it as process-fatal and a fresh process reopens the
// committed cursor (if any) only after lease expiry.
type unknownBulkOutcomeError struct{ error }

func isUnknownBulkOutcomeError(err error) bool {
	var target unknownBulkOutcomeError
	return errors.As(err, &target)
}