package records

import (
	"context"
	"errors"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/nexuscrm/kernel/pkg/logging"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/storage"
)

func TestBulkIngestContractCreateJobOwnsPermissionPreflightAndDefaults(t *testing.T) {
	ctx := context.Background()
	now := time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC)
	repo := newBulkIngestRepo()
	perms := &bulkIngestPerms{
		allow: map[string]bool{
			"Account:" + string(constants.PermCreate): true,
			"Account:" + string(constants.PermEdit):   false,
		},
		deferObjectGate: true,
	}
	adapter := newTestBulkIngestContract(repo, &bulkIngestMetadataFake{
		schemas: map[string]*models.ObjectMetadata{"Account": {APIName: "Account"}},
	}, perms, now, "job-1", t.TempDir())

	job, err := adapter.CreateIngestJob(ctx, contract.BulkIngestCreateRequest{
		Object:              "Account",
		Operation:           models.OperationUpsert,
		ExternalIDFieldName: "External_Id__c",
		APIVersion:          62.0,
	}, &models.UserSession{ID: "005-user"})
	if err != nil {
		t.Fatalf("CreateIngestJob returned error: %v", err)
	}
	if job.ID != "job-1" || job.Object != "Account" || job.Operation != models.OperationUpsert ||
		job.State != models.JobStateOpen || job.ConcurrencyMode != models.ConcurrencyParallel ||
		job.ContentType != "CSV" || job.ColumnDelimiter != "COMMA" || job.LineEnding != "LF" ||
		job.ApiVersion != 62.0 || job.CreatedById != "005-user" || job.ExternalIdFieldName != "External_Id__c" {
		t.Fatalf("created job = %#v", job)
	}
	if !job.CreatedDate.Equal(now) || !job.SystemModstamp.Equal(now) {
		t.Fatalf("job timestamps = %s/%s, want %s", job.CreatedDate, job.SystemModstamp, now)
	}
	if _, checkedEdit := perms.checked["Account:"+string(constants.PermEdit)]; checkedEdit {
		t.Fatalf("edit permission was checked despite record-aware deferral: %#v", perms.checked)
	}
	if _, checkedCreate := perms.checked["Account:"+string(constants.PermCreate)]; !checkedCreate {
		t.Fatalf("upsert create preflight was not checked: %#v", perms.checked)
	}
	if repo.jobs["job-1"] == nil || repo.jobs["job-1"].ApiVersion != 62.0 {
		t.Fatalf("persisted job = %#v, want creation API version 62", repo.jobs["job-1"])
	}
	read, err := adapter.GetIngestJob(ctx, "job-1", &models.UserSession{ID: "005-user"})
	if err != nil || read.ApiVersion != 62.0 {
		t.Fatalf("read job = %#v, %v; want persisted API version 62", read, err)
	}
	closed, err := adapter.CloseIngestJob(ctx, "job-1", models.JobStateAborted, &models.UserSession{ID: "005-user"})
	if err != nil || closed.ApiVersion != 62.0 {
		t.Fatalf("closed job = %#v, %v; want persisted API version 62", closed, err)
	}
}

func TestBulkIngestContractRejectsListViewGenericJobs(t *testing.T) {
	ctx := context.Background()
	repo := newBulkIngestRepo()
	perms := &bulkIngestPerms{
		allow: map[string]bool{
			constants.TableListView + ":" + string(constants.PermCreate): true,
		},
	}
	adapter := newTestBulkIngestContract(repo, &bulkIngestMetadataFake{
		schemas: map[string]*models.ObjectMetadata{constants.TableListView: {APIName: constants.TableListView}},
	}, perms, time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC), "job-list-view", t.TempDir())

	_, err := adapter.CreateIngestJob(ctx, contract.BulkIngestCreateRequest{
		Object:    constants.TableListView,
		Operation: models.OperationInsert,
	}, &models.UserSession{ID: "005-owner"})
	var bulkErr *contract.BulkIngestError
	if !errors.As(err, &bulkErr) {
		t.Fatalf("CreateIngestJob(_System_ListView) error = %T %v, want BulkIngestError", err, err)
	}
	if bulkErr.Kind != contract.BulkIngestErrorInvalidOperation {
		t.Fatalf("BulkIngestError.Kind = %s, want %s", bulkErr.Kind, contract.BulkIngestErrorInvalidOperation)
	}
	if len(repo.jobs) != 0 {
		t.Fatalf("CreateIngestJob(_System_ListView) persisted jobs = %#v, want none", repo.jobs)
	}
	if len(perms.checked) != 0 {
		t.Fatalf("CreateIngestJob(_System_ListView) checked permissions = %#v, want fail before permission preflight", perms.checked)
	}
}

func TestBulkIngestContractDoesNotExposeQueryJobs(t *testing.T) {
	ctx := context.Background()
	repo := newBulkIngestRepo()
	repo.jobs["query-job"] = &models.BulkJob{
		ID: "query-job", Operation: models.OperationQuery, State: models.JobStateOpen, CreatedById: "005-owner",
	}
	adapter := newTestBulkIngestContract(repo, &bulkIngestMetadataFake{}, &bulkIngestPerms{superUser: true}, time.Now(), "batch-1", t.TempDir())

	_, err := adapter.GetIngestJob(ctx, "query-job", &models.UserSession{ID: "005-owner"})
	var bulkErr *contract.BulkIngestError
	if !errors.As(err, &bulkErr) || bulkErr.Kind != contract.BulkIngestErrorNotFound {
		t.Fatalf("GetIngestJob(query job) error = %T %v, want not found", err, err)
	}
	_, err = adapter.CloseIngestJob(ctx, "query-job", models.JobStateAborted, &models.UserSession{ID: "005-owner"})
	if !errors.As(err, &bulkErr) || bulkErr.Kind != contract.BulkIngestErrorNotFound {
		t.Fatalf("CloseIngestJob(query job) error = %T %v, want not found", err, err)
	}
	err = adapter.UploadIngestData(ctx, contract.BulkIngestUploadRequest{
		JobID: "query-job", Body: stringsReader("Id\n001\n"), ContentLength: 7,
	}, &models.UserSession{ID: "005-owner"})
	if !errors.As(err, &bulkErr) || bulkErr.Kind != contract.BulkIngestErrorNotFound {
		t.Fatalf("UploadIngestData(query job) error = %T %v, want not found", err, err)
	}
	if len(repo.batches) != 0 {
		t.Fatalf("UploadIngestData(query job) created batches = %#v", repo.batches)
	}
	if repo.jobs["query-job"].State != models.JobStateOpen {
		t.Fatalf("query job state = %s, want Open", repo.jobs["query-job"].State)
	}
}

func TestBulkIngestContractUploadStoresFileAndCreatesBatchBelowSeam(t *testing.T) {
	ctx := context.Background()
	now := time.Date(2026, 6, 12, 13, 0, 0, 0, time.UTC)
	repo := newBulkIngestRepo()
	repo.jobs["job-1"] = &models.BulkJob{ID: "job-1", Operation: models.OperationInsert, State: models.JobStateOpen, CreatedById: "005-user"}
	uploadDir := t.TempDir()
	adapter := newTestBulkIngestContract(repo, &bulkIngestMetadataFake{}, &bulkIngestPerms{}, now, "batch-1", uploadDir)

	err := adapter.UploadIngestData(ctx, contract.BulkIngestUploadRequest{
		JobID:         "job-1",
		Body:          stringsReader("Name\nAcme\n"),
		ContentLength: int64(len("Name\nAcme\n")),
	}, &models.UserSession{ID: "005-user"})
	if err != nil {
		t.Fatalf("UploadIngestData returned error: %v", err)
	}
	fileBytes, err := os.ReadFile(uploadDir + "/bulk-ingest/job-1/batch-1.csv")
	if err != nil {
		t.Fatalf("read uploaded file: %v", err)
	}
	if string(fileBytes) != "Name\nAcme\n" {
		t.Fatalf("uploaded file = %q", string(fileBytes))
	}
	batch := repo.batches["batch-1"]
	if batch == nil {
		t.Fatal("batch was not created")
	}
	if batch.JobID != "job-1" || batch.State != "Queued" || !batch.CreatedDate.Equal(now) || !batch.SystemModstamp.Equal(now) {
		t.Fatalf("batch = %#v", batch)
	}
	if batch.ObjectKey != "bulk-ingest/job-1/batch-1.csv" || batch.ContentLength != int64(len("Name\nAcme\n")) || batch.ContentSHA256 == "" || batch.StorageAuthorityDigest == "" {
		t.Fatalf("durable batch storage binding = %#v", batch)
	}
}

func TestBulkIngestContractUploadPropagatesContractReadError(t *testing.T) {
	ctx := context.Background()
	repo := newBulkIngestRepo()
	repo.jobs["job-1"] = &models.BulkJob{ID: "job-1", Operation: models.OperationInsert, State: models.JobStateOpen, CreatedById: "005-user"}
	uploadDir := t.TempDir()
	adapter := newTestBulkIngestContract(repo, &bulkIngestMetadataFake{}, &bulkIngestPerms{}, time.Now(), "batch-too-large", uploadDir)

	err := adapter.UploadIngestData(ctx, contract.BulkIngestUploadRequest{
		JobID: "job-1",
		Body: bulkIngestFailingReader{err: &contract.BulkIngestError{
			Kind:    contract.BulkIngestErrorRequestTooLarge,
			Message: "too large",
		}},
	}, &models.UserSession{ID: "005-user"})
	var bulkErr *contract.BulkIngestError
	if !errors.As(err, &bulkErr) {
		t.Fatalf("UploadIngestData error = %T %[1]v, want BulkIngestError", err)
	}
	if bulkErr.Kind != contract.BulkIngestErrorRequestTooLarge {
		t.Fatalf("BulkIngestError.Kind = %s, want %s", bulkErr.Kind, contract.BulkIngestErrorRequestTooLarge)
	}
	if _, statErr := os.Stat(uploadDir + "/bulk-ingest/job-1/batch-too-large.csv"); !os.IsNotExist(statErr) {
		t.Fatalf("partial upload file stat err = %v, want not exist", statErr)
	}
	if len(repo.batches) != 0 {
		t.Fatalf("batches = %#v, want none", repo.batches)
	}
}

func TestBulkIngestContractUploadLosingCloseRaceDeletesUnacceptedObject(t *testing.T) {
	repo := newBulkIngestRepo()
	repo.jobs["job-1"] = &models.BulkJob{ID: "job-1", Operation: models.OperationInsert, State: models.JobStateOpen, CreatedById: "005-user"}
	repo.closeBeforeBatch = true
	uploadDir := t.TempDir()
	adapter := newTestBulkIngestContract(repo, &bulkIngestMetadataFake{}, &bulkIngestPerms{}, time.Now(), "batch-race", uploadDir)

	err := adapter.UploadIngestData(t.Context(), contract.BulkIngestUploadRequest{
		JobID: "job-1", Body: stringsReader("Name\nAcme\n"), ContentLength: int64(len("Name\nAcme\n")),
	}, &models.UserSession{ID: "005-user"})
	var bulkErr *contract.BulkIngestError
	if !errors.As(err, &bulkErr) || bulkErr.Kind != contract.BulkIngestErrorInvalidJobState {
		t.Fatalf("UploadIngestData race error = %T %v", err, err)
	}
	if _, statErr := os.Stat(uploadDir + "/bulk-ingest/job-1/batch-race.csv"); !os.IsNotExist(statErr) {
		t.Fatalf("unaccepted object remains after close race: %v", statErr)
	}
	if len(repo.batches) != 0 {
		t.Fatalf("close-race upload registered batches: %#v", repo.batches)
	}
}

func TestBulkIngestContractUploadRetainsObjectWhenBatchAdmissionOutcomeIsUnknown(t *testing.T) {
	repo := newBulkIngestRepo()
	repo.jobs["job-1"] = &models.BulkJob{ID: "job-1", Operation: models.OperationInsert, State: models.JobStateOpen, CreatedById: "005-user"}
	repo.admissionOutcome = ports.BulkBatchAdmissionUnknown
	repo.admissionErr = errors.New("commit acknowledgement lost")
	uploadDir := t.TempDir()
	adapter := newTestBulkIngestContract(repo, &bulkIngestMetadataFake{}, &bulkIngestPerms{}, time.Now(), "batch-unknown", uploadDir)

	err := adapter.UploadIngestData(t.Context(), contract.BulkIngestUploadRequest{
		JobID: "job-1", Body: stringsReader("Name\nAcme\n"), ContentLength: int64(len("Name\nAcme\n")),
	}, &models.UserSession{ID: "005-user"})
	if err == nil {
		t.Fatal("UploadIngestData outcome-unknown admission unexpectedly succeeded")
	}
	if _, statErr := os.Stat(uploadDir + "/bulk-ingest/job-1/batch-unknown.csv"); statErr != nil {
		t.Fatalf("outcome-unknown admission deleted the only uploaded object: %v", statErr)
	}
}

func TestBulkIngestContractUploadDeletesObjectAfterProvenNonAdmission(t *testing.T) {
	repo := newBulkIngestRepo()
	repo.jobs["job-1"] = &models.BulkJob{ID: "job-1", Operation: models.OperationInsert, State: models.JobStateOpen, CreatedById: "005-user"}
	repo.admissionOutcome = ports.BulkBatchAdmissionNotCreated
	repo.admissionErr = errors.New("insert rolled back")
	uploadDir := t.TempDir()
	adapter := newTestBulkIngestContract(repo, &bulkIngestMetadataFake{}, &bulkIngestPerms{}, time.Now(), "batch-rolled-back", uploadDir)

	err := adapter.UploadIngestData(t.Context(), contract.BulkIngestUploadRequest{
		JobID: "job-1", Body: stringsReader("Name\nAcme\n"), ContentLength: int64(len("Name\nAcme\n")),
	}, &models.UserSession{ID: "005-user"})
	if err == nil {
		t.Fatal("UploadIngestData rolled-back admission unexpectedly succeeded")
	}
	if _, statErr := os.Stat(uploadDir + "/bulk-ingest/job-1/batch-rolled-back.csv"); !os.IsNotExist(statErr) {
		t.Fatalf("proven-unaccepted object remains: %v", statErr)
	}
}

func TestBulkIngestContractCloseDurablyAcceptsWithoutEventAuthority(t *testing.T) {
	ctx := context.Background()
	repo := newBulkIngestRepo()
	repo.jobs["job-1"] = &models.BulkJob{ID: "job-1", Operation: models.OperationInsert, State: models.JobStateOpen, CreatedById: "005-user"}
	adapter := newTestBulkIngestContract(repo, &bulkIngestMetadataFake{}, &bulkIngestPerms{}, time.Now(), "unused", t.TempDir())

	job, err := adapter.CloseIngestJob(
		ctx,
		"job-1",
		models.JobStateUploadComplete,
		&models.UserSession{
			ID:           "005-user",
			SessionClass: "interactive",
		},
	)
	if err != nil {
		t.Fatalf("CloseIngestJob returned error: %v", err)
	}
	if job.State != models.JobStateUploadComplete || repo.jobs["job-1"].State != models.JobStateUploadComplete {
		t.Fatalf("job state = %s repo=%s", job.State, repo.jobs["job-1"].State)
	}
}

func TestBulkIngestContractCloseRequiresActiveIndependentWorker(t *testing.T) {
	repo := newBulkIngestRepo()
	repo.jobs["job-1"] = &models.BulkJob{ID: "job-1", Operation: models.OperationInsert, State: models.JobStateOpen, CreatedById: "005-user"}
	repo.workerErr = errors.New("no active Bulk owner")
	adapter := newTestBulkIngestContract(repo, &bulkIngestMetadataFake{}, &bulkIngestPerms{}, time.Now(), "unused", t.TempDir())

	_, err := adapter.CloseIngestJob(t.Context(), "job-1", models.JobStateUploadComplete, &models.UserSession{ID: "005-user"})
	if err == nil || !strings.Contains(err.Error(), "worker is not ready") {
		t.Fatalf("CloseIngestJob error = %v", err)
	}
	if repo.jobs["job-1"].State != models.JobStateOpen {
		t.Fatalf("unowned work was accepted: state=%s", repo.jobs["job-1"].State)
	}
}

func TestBulkIngestContractDuplicateCloseRemainsIdempotentWhenWorkerLaterUnavailable(t *testing.T) {
	repo := newBulkIngestRepo()
	repo.jobs["job-1"] = &models.BulkJob{ID: "job-1", Operation: models.OperationInsert, State: models.JobStateOpen, CreatedById: "005-user"}
	adapter := newTestBulkIngestContract(repo, &bulkIngestMetadataFake{}, &bulkIngestPerms{}, time.Now(), "unused", t.TempDir())
	job, err := adapter.CloseIngestJob(t.Context(), "job-1", models.JobStateUploadComplete, &models.UserSession{ID: "005-user"})
	if err != nil || job.State != models.JobStateUploadComplete {
		t.Fatalf("first close = %#v, %v", job, err)
	}

	repo.workerErr = errors.New("worker became unavailable")
	job, err = adapter.CloseIngestJob(t.Context(), "job-1", models.JobStateUploadComplete, &models.UserSession{ID: "005-user"})
	if err != nil || job.State != models.JobStateUploadComplete {
		t.Fatalf("idempotent close while worker unavailable = %#v, %v", job, err)
	}
}

func TestBulkIngestContractAbortOwnsEveryNonterminalState(t *testing.T) {
	for _, initial := range []models.BulkJobState{
		models.JobStateOpen,
		models.JobStateUploadComplete,
		models.JobStateInProgress,
	} {
		t.Run(string(initial), func(t *testing.T) {
			repo := newBulkIngestRepo()
			repo.jobs["job-1"] = &models.BulkJob{ID: "job-1", Operation: models.OperationInsert, State: initial, CreatedById: "005-user"}
			adapter := newTestBulkIngestContract(repo, &bulkIngestMetadataFake{}, &bulkIngestPerms{}, time.Now(), "unused", t.TempDir())

			job, err := adapter.CloseIngestJob(
				t.Context(), "job-1", models.JobStateAborted, &models.UserSession{ID: "005-user"},
			)
			if err != nil || job.State != models.JobStateAborted || repo.jobs["job-1"].State != models.JobStateAborted {
				t.Fatalf("abort from %s = %#v repo=%#v err=%v", initial, job, repo.jobs["job-1"], err)
			}
		})
	}
}

func TestBulkIngestContractAbortIsIdempotentButCannotRewriteTerminalOutcome(t *testing.T) {
	t.Run("duplicate abort", func(t *testing.T) {
		repo := newBulkIngestRepo()
		repo.jobs["job-1"] = &models.BulkJob{ID: "job-1", Operation: models.OperationInsert, State: models.JobStateAborted, CreatedById: "005-user"}
		adapter := newTestBulkIngestContract(repo, &bulkIngestMetadataFake{}, &bulkIngestPerms{}, time.Now(), "unused", t.TempDir())
		job, err := adapter.CloseIngestJob(t.Context(), "job-1", models.JobStateAborted, &models.UserSession{ID: "005-user"})
		if err != nil || job.State != models.JobStateAborted {
			t.Fatalf("duplicate abort = %#v, %v", job, err)
		}
	})

	for _, terminal := range []models.BulkJobState{models.JobStateJobComplete, models.JobStateFailed} {
		t.Run(string(terminal), func(t *testing.T) {
			repo := newBulkIngestRepo()
			repo.jobs["job-1"] = &models.BulkJob{ID: "job-1", Operation: models.OperationInsert, State: terminal, CreatedById: "005-user"}
			adapter := newTestBulkIngestContract(repo, &bulkIngestMetadataFake{}, &bulkIngestPerms{}, time.Now(), "unused", t.TempDir())
			_, err := adapter.CloseIngestJob(t.Context(), "job-1", models.JobStateAborted, &models.UserSession{ID: "005-user"})
			var bulkErr *contract.BulkIngestError
			if !errors.As(err, &bulkErr) || bulkErr.Kind != contract.BulkIngestErrorInvalidJobState || repo.jobs["job-1"].State != terminal {
				t.Fatalf("abort terminal %s error=%T %v state=%s", terminal, err, err, repo.jobs["job-1"].State)
			}
		})
	}
}

func TestBulkIngestContractSuperUserClosePreservesDurableOwner(
	t *testing.T,
) {
	ctx := context.Background()
	repo := newBulkIngestRepo()
	repo.jobs["job-1"] = &models.BulkJob{
		ID:          "job-1",
		Operation:   models.OperationInsert,
		State:       models.JobStateOpen,
		CreatedById: "005-owner",
	}
	adapter := newTestBulkIngestContract(
		repo,
		&bulkIngestMetadataFake{},
		&bulkIngestPerms{superUser: true},
		time.Now(),
		"unused",
		t.TempDir(),
	)

	_, err := adapter.CloseIngestJob(
		ctx,
		"job-1",
		models.JobStateUploadComplete,
		&models.UserSession{
			ID:           "005-admin",
			SessionClass: "platform_browser",
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	if repo.jobs["job-1"].CreatedById != "005-owner" {
		t.Fatalf(
			"delegated close changed job owner to %q",
			repo.jobs["job-1"].CreatedById,
		)
	}
}

func newTestBulkIngestContract(repo *bulkIngestRepo, metadataSource bulkIngestMetadata, perms bulkIngestPermissions, now time.Time, id string, uploadDir string) contract.BulkIngest {
	config := storage.Config{Provider: storage.ProviderFilesystem, FilesystemRoot: uploadDir, Bucket: storage.DefaultFilesystemBucket}
	store, err := storage.NewStore(config)
	if err != nil {
		panic(err)
	}
	digest, err := storage.ConfigAuthorityDigest(config)
	if err != nil {
		panic(err)
	}
	return &bulkIngestContract{
		jobs:          NewBulkService(repo, logging.New(context.Background())),
		metadata:      metadataSource,
		permissions:   perms,
		now:           func() time.Time { return now },
		idgen:         func() string { return id },
		objectStore:   store,
		storage:       config,
		storageDigest: digest,
	}
}

type bulkIngestRepo struct {
	jobs             map[string]*models.BulkJob
	batches          map[string]*models.BulkBatch
	workerErr        error
	closeBeforeBatch bool
	admissionOutcome ports.BulkBatchAdmissionOutcome
	admissionErr     error
}

func newBulkIngestRepo() *bulkIngestRepo {
	return &bulkIngestRepo{
		jobs:    make(map[string]*models.BulkJob),
		batches: make(map[string]*models.BulkBatch),
	}
}

func (r *bulkIngestRepo) CreateJob(_ context.Context, job *models.BulkJob) error {
	r.jobs[job.ID] = cloneBulkIngestJob(job)
	return nil
}

func (r *bulkIngestRepo) GetJob(_ context.Context, jobID string) (*models.BulkJob, error) {
	return cloneBulkIngestJob(r.jobs[jobID]), nil
}

func (r *bulkIngestRepo) UpdateJob(_ context.Context, job *models.BulkJob) error {
	r.jobs[job.ID] = cloneBulkIngestJob(job)
	return nil
}

func (r *bulkIngestRepo) CreateBatchIfJobOpen(_ context.Context, batch *models.BulkBatch) (ports.BulkBatchAdmissionOutcome, error) {
	if r.admissionErr != nil {
		outcome := r.admissionOutcome
		if outcome == "" {
			outcome = ports.BulkBatchAdmissionUnknown
		}
		return outcome, r.admissionErr
	}
	job := r.jobs[batch.JobID]
	if r.closeBeforeBatch && job != nil {
		job.State = models.JobStateUploadComplete
	}
	if job == nil || job.State != models.JobStateOpen {
		return ports.BulkBatchAdmissionNotCreated, nil
	}
	cp := *batch
	r.batches[batch.ID] = &cp
	return ports.BulkBatchAdmissionCreated, nil
}

func (r *bulkIngestRepo) AcceptIngestJob(_ context.Context, jobID string) (bool, error) {
	job := r.jobs[jobID]
	if job == nil || job.State != models.JobStateOpen {
		return false, nil
	}
	job.State = models.JobStateUploadComplete
	return true, nil
}

func (r *bulkIngestRepo) AbortIngestJob(_ context.Context, jobID string) (bool, error) {
	job := r.jobs[jobID]
	if job == nil || (job.State != models.JobStateOpen && job.State != models.JobStateUploadComplete && job.State != models.JobStateInProgress) {
		return false, nil
	}
	job.State = models.JobStateAborted
	return true, nil
}

func (r *bulkIngestRepo) VerifyActiveWorker(context.Context, string) error { return r.workerErr }

func (r *bulkIngestRepo) GetBatches(context.Context, string) ([]*models.BulkBatch, error) {
	return nil, nil
}

func cloneBulkIngestJob(job *models.BulkJob) *models.BulkJob {
	if job == nil {
		return nil
	}
	cp := *job
	return &cp
}

type bulkIngestMetadataFake struct {
	schemas map[string]*models.ObjectMetadata
}

func (m *bulkIngestMetadataFake) GetSchema(_ context.Context, objectAPIName string) *models.ObjectMetadata {
	return m.schemas[objectAPIName]
}

type bulkIngestPerms struct {
	allow           map[string]bool
	checked         map[string]bool
	deferObjectGate bool
	superUser       bool
}

func (p *bulkIngestPerms) CheckObjectPermissionWithUser(_ context.Context, objectAPIName string, operation string, _ *models.UserSession) bool {
	if p.checked == nil {
		p.checked = make(map[string]bool)
	}
	key := objectAPIName + ":" + operation
	p.checked[key] = true
	return p.allow[key]
}

func (p *bulkIngestPerms) RequiresRecordAwareCRUDDeferral(context.Context, *models.ObjectMetadata, string) bool {
	return p.deferObjectGate
}

func (p *bulkIngestPerms) HasSuperUserCapability(context.Context, *models.UserSession) bool {
	return p.superUser
}

func stringsReader(s string) *strings.Reader {
	return strings.NewReader(s)
}

type bulkIngestFailingReader struct {
	err error
}

func (r bulkIngestFailingReader) Read([]byte) (int, error) {
	return 0, r.err
}