package files

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

	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/storage"
)

type durableUploadTestJournal struct {
	row         persistence.FileUploadOperation
	acceptCalls int
	loadCalls   int
}

type completionRaceUploadJournal struct {
	*durableUploadTestJournal
	reloaded persistence.FileUploadOperation
}

func (j *completionRaceUploadJournal) Load(
	context.Context,
	string,
) (persistence.FileUploadOperation, error) {
	return j.reloaded, nil
}

func (*completionRaceUploadJournal) MarkCompleted(
	context.Context,
	string,
	string,
	string,
	string,
	time.Time,
) error {
	return persistence.ErrFileUploadOperationLostOwnership
}

func (*completionRaceUploadJournal) FinalizeBlobStored(
	context.Context,
	string,
	string,
	string,
	string,
	time.Time,
	func(context.Context) error,
) error {
	return persistence.ErrFileUploadOperationLostOwnership
}

func (j *durableUploadTestJournal) Accept(
	_ context.Context,
	input persistence.FileUploadOperationAcceptance,
) (persistence.FileUploadOperation, bool, error) {
	j.acceptCalls++
	if j.row.ID == "" {
		j.row = persistence.FileUploadOperation{
			ID: input.ID, IdempotencyKey: input.IdempotencyKey, RequestHash: input.RequestHash,
			State: persistence.FileUploadOperationPrepared, OwnerToken: input.OwnerToken, ObjectKey: input.ObjectKey,
			StorageProvider: input.StorageProvider, StorageBucket: input.StorageBucket,
			StorageAuthorityDigest: input.StorageAuthorityDigest,
			SystemFileID:           input.SystemFileID, ParentID: input.ParentID,
			ParentObjectAPIName: input.ParentObjectAPIName, ExpiresAt: input.LeaseUntil,
		}
		return j.row, true, nil
	}
	if j.row.ID != input.ID || j.row.IdempotencyKey != input.IdempotencyKey ||
		(j.row.RequestHash != input.RequestHash &&
			!strings.HasPrefix(j.row.RequestHash, input.RequestHash+"|")) ||
		j.row.SystemFileID != input.SystemFileID ||
		j.row.StorageProvider != input.StorageProvider ||
		j.row.StorageBucket != input.StorageBucket ||
		j.row.StorageAuthorityDigest != input.StorageAuthorityDigest ||
		j.row.ParentID != input.ParentID || j.row.ParentObjectAPIName != input.ParentObjectAPIName {
		return j.row, false, persistence.ErrFileUploadOperationConflict
	}
	return j.row, false, nil
}

func (j *durableUploadTestJournal) Load(
	context.Context,
	string,
) (persistence.FileUploadOperation, error) {
	j.loadCalls++
	if j.row.ID == "" {
		return persistence.FileUploadOperation{}, persistence.ErrFileUploadOperationNotFound
	}
	return j.row, nil
}

func (j *durableUploadTestJournal) MarkBlobStored(
	_ context.Context,
	id string,
	ownerToken string,
	headerHash string,
	completeHash string,
	objectKey string,
	expiresAt time.Time,
) error {
	if j.row.ID != id || j.row.State != persistence.FileUploadOperationPrepared ||
		j.row.OwnerToken != ownerToken || j.row.RequestHash != headerHash ||
		j.row.ObjectKey != objectKey {
		return persistence.ErrFileUploadOperationLostOwnership
	}
	j.row.State = persistence.FileUploadOperationBlobStored
	j.row.RequestHash = completeHash
	j.row.ExpiresAt = expiresAt
	return nil
}

func (j *durableUploadTestJournal) MarkCompleted(
	_ context.Context,
	id string,
	ownerToken string,
	completeHash string,
	objectKey string,
	expiresAt time.Time,
) error {
	if j.row.ID != id || j.row.State != persistence.FileUploadOperationBlobStored ||
		j.row.OwnerToken != ownerToken || j.row.RequestHash != completeHash ||
		j.row.ObjectKey != objectKey {
		return persistence.ErrFileUploadOperationLostOwnership
	}
	j.row.State = persistence.FileUploadOperationCompleted
	j.row.ExpiresAt = expiresAt
	return nil
}

func (j *durableUploadTestJournal) FinalizeBlobStored(
	ctx context.Context,
	id string,
	ownerToken string,
	completeHash string,
	objectKey string,
	expiresAt time.Time,
	commitMetadata func(context.Context) error,
) error {
	if j.row.ID != id || j.row.State != persistence.FileUploadOperationBlobStored ||
		j.row.OwnerToken != ownerToken || j.row.RequestHash != completeHash ||
		j.row.ObjectKey != objectKey {
		return persistence.ErrFileUploadOperationLostOwnership
	}
	if err := commitMetadata(ctx); err != nil {
		return err
	}
	j.row.State = persistence.FileUploadOperationCompleted
	j.row.ExpiresAt = expiresAt
	return nil
}

type durableUploadTestObjectStore struct {
	objects map[string]string
	puts    int
	gets    int
	deletes int
}

type fencedUploadTestJournal struct {
	mu  sync.Mutex
	row persistence.FileUploadOperation
}

func (j *fencedUploadTestJournal) Accept(
	_ context.Context,
	input persistence.FileUploadOperationAcceptance,
) (persistence.FileUploadOperation, bool, error) {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.row.ID == "" {
		j.row = persistence.FileUploadOperation{
			ID: input.ID, IdempotencyKey: input.IdempotencyKey, RequestHash: input.RequestHash,
			State: persistence.FileUploadOperationPrepared, OwnerToken: input.OwnerToken,
			ObjectKey: input.ObjectKey, SystemFileID: input.SystemFileID,
			StorageProvider: input.StorageProvider, StorageBucket: input.StorageBucket,
			StorageAuthorityDigest: input.StorageAuthorityDigest,
			ParentID:               input.ParentID, ParentObjectAPIName: input.ParentObjectAPIName,
			ExpiresAt: input.LeaseUntil,
		}
		return j.row, true, nil
	}
	return j.row, false, nil
}

func (j *fencedUploadTestJournal) Load(
	context.Context,
	string,
) (persistence.FileUploadOperation, error) {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.row.ID == "" {
		return persistence.FileUploadOperation{}, persistence.ErrFileUploadOperationNotFound
	}
	return j.row, nil
}

func (j *fencedUploadTestJournal) MarkBlobStored(
	_ context.Context,
	id string,
	ownerToken string,
	headerHash string,
	completeHash string,
	objectKey string,
	expiresAt time.Time,
) error {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.row.ID != id || j.row.State != persistence.FileUploadOperationPrepared ||
		j.row.OwnerToken != ownerToken || j.row.RequestHash != headerHash ||
		j.row.ObjectKey != objectKey {
		return persistence.ErrFileUploadOperationLostOwnership
	}
	j.row.State = persistence.FileUploadOperationBlobStored
	j.row.RequestHash = completeHash
	j.row.ExpiresAt = expiresAt
	return nil
}

func (j *fencedUploadTestJournal) MarkCompleted(
	_ context.Context,
	id string,
	ownerToken string,
	completeHash string,
	objectKey string,
	expiresAt time.Time,
) error {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.row.ID != id || j.row.State != persistence.FileUploadOperationBlobStored ||
		j.row.OwnerToken != ownerToken || j.row.RequestHash != completeHash ||
		j.row.ObjectKey != objectKey {
		return persistence.ErrFileUploadOperationLostOwnership
	}
	j.row.State = persistence.FileUploadOperationCompleted
	j.row.ExpiresAt = expiresAt
	return nil
}

func (j *fencedUploadTestJournal) FinalizeBlobStored(
	ctx context.Context,
	id string,
	ownerToken string,
	completeHash string,
	objectKey string,
	expiresAt time.Time,
	commitMetadata func(context.Context) error,
) error {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.row.ID != id || j.row.State != persistence.FileUploadOperationBlobStored ||
		j.row.OwnerToken != ownerToken || j.row.RequestHash != completeHash ||
		j.row.ObjectKey != objectKey {
		return persistence.ErrFileUploadOperationLostOwnership
	}
	if err := commitMetadata(ctx); err != nil {
		return err
	}
	j.row.State = persistence.FileUploadOperationCompleted
	j.row.ExpiresAt = expiresAt
	return nil
}

type blockingFirstUploadObjectStore struct {
	mu           sync.Mutex
	objects      map[string]string
	firstStarted chan struct{}
	releaseFirst chan struct{}
	puts         int
	deleteErrors int
}

func (s *blockingFirstUploadObjectStore) PutObject(
	_ context.Context,
	key string,
	body io.Reader,
	options storage.PutObjectOptions,
) (storage.ObjectMetadata, error) {
	value, err := io.ReadAll(body)
	if err != nil {
		return storage.ObjectMetadata{}, err
	}
	s.mu.Lock()
	s.puts++
	first := s.puts == 1
	s.mu.Unlock()
	if first {
		close(s.firstStarted)
		<-s.releaseFirst
	}
	s.mu.Lock()
	if s.objects == nil {
		s.objects = map[string]string{}
	}
	s.objects[key] = string(value)
	s.mu.Unlock()
	return storage.ObjectMetadata{
		Key: key, Size: int64(len(value)), ContentType: options.ContentType,
	}, nil
}

func (s *blockingFirstUploadObjectStore) GetObject(
	_ context.Context,
	key string,
) (io.ReadCloser, storage.ObjectMetadata, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	value, ok := s.objects[key]
	if !ok {
		return nil, storage.ObjectMetadata{}, storage.ErrObjectNotFound
	}
	return io.NopCloser(strings.NewReader(value)), storage.ObjectMetadata{
		Key: key, Size: int64(len(value)), ContentType: "text/plain",
	}, nil
}

func (s *blockingFirstUploadObjectStore) HeadObject(
	_ context.Context,
	key string,
) (storage.ObjectMetadata, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	value, ok := s.objects[key]
	if !ok {
		return storage.ObjectMetadata{}, storage.ErrObjectNotFound
	}
	return storage.ObjectMetadata{Key: key, Size: int64(len(value)), ContentType: "text/plain"}, nil
}

func (s *blockingFirstUploadObjectStore) DeleteObject(_ context.Context, key string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.deleteErrors > 0 {
		s.deleteErrors--
		return errors.New("delete attempt object")
	}
	delete(s.objects, key)
	return nil
}

func (*blockingFirstUploadObjectStore) GeneratePresignedGetURL(
	context.Context,
	string,
	time.Duration,
) (string, error) {
	return "", storage.ErrPresignUnsupported
}

func (s *durableUploadTestObjectStore) PutObject(
	_ context.Context,
	key string,
	body io.Reader,
	options storage.PutObjectOptions,
) (storage.ObjectMetadata, error) {
	value, err := io.ReadAll(body)
	if err != nil {
		return storage.ObjectMetadata{}, err
	}
	if s.objects == nil {
		s.objects = map[string]string{}
	}
	s.objects[key] = string(value)
	s.puts++
	return storage.ObjectMetadata{
		Key: key, Size: int64(len(value)), ContentType: options.ContentType,
	}, nil
}

func (s *durableUploadTestObjectStore) GetObject(
	_ context.Context,
	key string,
) (io.ReadCloser, storage.ObjectMetadata, error) {
	s.gets++
	value, ok := s.objects[key]
	if !ok {
		return nil, storage.ObjectMetadata{}, storage.ErrObjectNotFound
	}
	return io.NopCloser(strings.NewReader(value)), storage.ObjectMetadata{
		Key: key, Size: int64(len(value)), ContentType: "text/plain",
	}, nil
}

func (s *durableUploadTestObjectStore) HeadObject(
	_ context.Context,
	key string,
) (storage.ObjectMetadata, error) {
	value, ok := s.objects[key]
	if !ok {
		return storage.ObjectMetadata{}, storage.ErrObjectNotFound
	}
	return storage.ObjectMetadata{Key: key, Size: int64(len(value)), ContentType: "text/plain"}, nil
}

func (s *durableUploadTestObjectStore) DeleteObject(_ context.Context, key string) error {
	delete(s.objects, key)
	s.deletes++
	return nil
}

func (*durableUploadTestObjectStore) GeneratePresignedGetURL(
	context.Context,
	string,
	time.Duration,
) (string, error) {
	return "", storage.ErrPresignUnsupported
}

type firstMarkBlobStoredErrorJournal struct {
	*fencedUploadTestJournal
	err       error
	committed bool
	calls     int
}

func (j *firstMarkBlobStoredErrorJournal) MarkBlobStored(
	ctx context.Context,
	id string,
	ownerToken string,
	headerHash string,
	completeHash string,
	objectKey string,
	expiresAt time.Time,
) error {
	j.calls++
	if j.calls != 1 {
		return j.fencedUploadTestJournal.MarkBlobStored(
			ctx, id, ownerToken, headerHash, completeHash, objectKey, expiresAt,
		)
	}
	if j.committed {
		if err := j.fencedUploadTestJournal.MarkBlobStored(
			ctx, id, ownerToken, headerHash, completeHash, objectKey, expiresAt,
		); err != nil {
			return err
		}
	}
	return j.err
}

type firstPutOutcomeObjectStore struct {
	durableUploadTestObjectStore
	err       error
	wrongSize bool
}

func (s *firstPutOutcomeObjectStore) PutObject(
	ctx context.Context,
	key string,
	body io.Reader,
	options storage.PutObjectOptions,
) (storage.ObjectMetadata, error) {
	meta, err := s.durableUploadTestObjectStore.PutObject(ctx, key, body, options)
	if err != nil || s.puts != 1 {
		return meta, err
	}
	if s.wrongSize {
		meta.Size++
	}
	return meta, s.err
}

func TestDurableRecordUploadReplaysAfterResponseLossAndRestart(t *testing.T) {
	journal := &durableUploadTestJournal{}
	objectStore := &durableUploadTestObjectStore{}
	records := &captureFileRecordStore{}
	service := newDurableUploadTestService(objectStore, records, journal)
	request := durableUploadTestRequest("same")

	first, err := service.StoreRecordAttachment(t.Context(), request)
	if err != nil {
		t.Fatal(err)
	}
	if journal.row.State != persistence.FileUploadOperationCompleted || objectStore.puts != 1 {
		t.Fatalf("first upload = state %q puts %d", journal.row.State, objectStore.puts)
	}

	// A new service instance represents a process restart after the first
	// response was lost. It must replay the durable result without another Put.
	restarted := newDurableUploadTestService(objectStore, records, journal)
	request.Body = strings.NewReader("same")
	second, err := restarted.StoreRecordAttachment(t.Context(), request)
	if err != nil {
		t.Fatal(err)
	}
	if first.ID != second.ID || objectStore.puts != 1 || journal.acceptCalls != 1 {
		t.Fatalf(
			"restart replay = first %#v second %#v puts %d accepts %d",
			first, second, objectStore.puts, journal.acceptCalls,
		)
	}

	request.Body = strings.NewReader("diff")
	if _, err := restarted.StoreRecordAttachment(t.Context(), request); err == nil {
		t.Fatal("completed upload accepted different replay content")
	}
	if objectStore.puts != 1 {
		t.Fatalf("conflicting replay wrote object %d times", objectStore.puts)
	}
}

func TestDurableRecordUploadExistingMetadataRequiresJournalStorageAuthority(t *testing.T) {
	for _, state := range []string{
		persistence.FileUploadOperationBlobStored,
		persistence.FileUploadOperationCompleted,
	} {
		for _, mutate := range []struct {
			name  string
			apply func(models.SObject)
		}{
			{
				name: "provider mismatch",
				apply: func(record models.SObject) {
					record[constants.FieldSysFile_StorageProvider] = storage.ProviderS3
				},
			},
			{
				name: "bucket mismatch",
				apply: func(record models.SObject) {
					record[constants.FieldSysFile_StorageBucket] = "other-bucket"
				},
			},
		} {
			t.Run(state+"/"+mutate.name, func(t *testing.T) {
				journal := &durableUploadTestJournal{}
				objectStore := &durableUploadTestObjectStore{}
				records := &captureFileRecordStore{}
				service := newDurableUploadTestService(objectStore, records, journal)
				request := durableUploadTestRequest("same")
				if _, err := service.StoreRecordAttachment(t.Context(), request); err != nil {
					t.Fatal(err)
				}
				journal.row.State = state
				mutate.apply(records.inserted)
				request.Body = strings.NewReader("same")
				if _, err := service.StoreRecordAttachment(t.Context(), request); err == nil {
					t.Fatal("mixed journal/metadata storage authority was accepted")
				}
				if journal.row.State != state {
					t.Fatalf(
						"mixed storage authority changed journal state to %q",
						journal.row.State,
					)
				}
				if objectStore.puts != 1 {
					t.Fatalf("mixed storage authority wrote %d objects", objectStore.puts)
				}
			})
		}
	}
}

func TestDurableRecordUploadDoesNotLoseAttemptObjectAfterPrecommitFailure(t *testing.T) {
	tests := []struct {
		name  string
		store *firstPutOutcomeObjectStore
	}{
		{
			name: "object store committed but returned an error",
			store: &firstPutOutcomeObjectStore{
				err: errors.New("object store response lost"),
			},
		},
		{
			name:  "object store committed an unexpected size receipt",
			store: &firstPutOutcomeObjectStore{wrongSize: true},
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			journal := &fencedUploadTestJournal{}
			records := &captureFileRecordStore{}
			service := newDurableUploadTestService(test.store, records, journal)

			if _, err := service.StoreRecordAttachment(
				t.Context(), durableUploadTestRequest("same"),
			); err == nil {
				t.Fatal("first upload unexpectedly succeeded")
			}
			if len(test.store.objects) != 1 {
				t.Fatalf("first upload objects = %#v, want one committed attempt", test.store.objects)
			}
			firstOwner, firstKey := journal.row.OwnerToken, journal.row.ObjectKey

			if _, err := service.StoreRecordAttachment(
				t.Context(), durableUploadTestRequest("same"),
			); err == nil {
				t.Fatal("retry rotated an uncleared attempt")
			} else {
				var inProgress *uploadInProgressError
				if !errors.As(err, &inProgress) {
					t.Fatalf("retry error = %v, want upload in progress", err)
				}
			}
			if len(test.store.objects) != 1 || test.store.puts != 1 ||
				journal.row.OwnerToken != firstOwner || journal.row.ObjectKey != firstKey {
				t.Fatalf(
					"retry changed uncleared attempt: row %#v puts %d objects %#v",
					journal.row, test.store.puts, test.store.objects,
				)
			}
		})
	}
}

func TestDurableRecordUploadDoesNotLoseAttemptObjectAfterJournalFailure(t *testing.T) {
	tests := []struct {
		name      string
		committed bool
	}{
		{name: "journal update did not commit"},
		{name: "journal update committed but response was lost", committed: true},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			journal := &firstMarkBlobStoredErrorJournal{
				fencedUploadTestJournal: &fencedUploadTestJournal{},
				err:                     errors.New("journal response lost"),
				committed:               test.committed,
			}
			objectStore := &durableUploadTestObjectStore{}
			records := &captureFileRecordStore{}
			service := newDurableUploadTestService(objectStore, records, journal)

			if _, err := service.StoreRecordAttachment(
				t.Context(), durableUploadTestRequest("same"),
			); err == nil {
				t.Fatal("first upload unexpectedly succeeded")
			}
			_, retryErr := service.StoreRecordAttachment(
				t.Context(), durableUploadTestRequest("same"),
			)
			if test.committed {
				if retryErr != nil {
					t.Fatalf("committed response-loss retry: %v", retryErr)
				}
			} else {
				var inProgress *uploadInProgressError
				if !errors.As(retryErr, &inProgress) {
					t.Fatalf("uncommitted retry error = %v, want upload in progress", retryErr)
				}
			}
			if len(objectStore.objects) != 1 || objectStore.puts != 1 {
				t.Fatalf("restart changed attempt objects: puts %d objects %#v",
					objectStore.puts, objectStore.objects)
			}
		})
	}
}

func TestDurableRecordUploadConcurrentRetryCannotRotateInFlightAttempt(t *testing.T) {
	journal := &fencedUploadTestJournal{}
	objectStore := &blockingFirstUploadObjectStore{
		objects:      map[string]string{},
		firstStarted: make(chan struct{}),
		releaseFirst: make(chan struct{}),
	}
	records := &captureFileRecordStore{}
	firstService := newDurableUploadTestService(objectStore, records, journal)
	secondService := newDurableUploadTestService(objectStore, records, journal)

	firstErr := make(chan error, 1)
	go func() {
		_, err := firstService.StoreRecordAttachment(
			context.Background(), durableUploadTestRequest("same"),
		)
		firstErr <- err
	}()
	<-objectStore.firstStarted

	if _, err := secondService.StoreRecordAttachment(
		t.Context(), durableUploadTestRequest("same"),
	); err == nil {
		t.Fatal("concurrent retry rotated an in-flight attempt")
	} else {
		var inProgress *uploadInProgressError
		if !errors.As(err, &inProgress) {
			t.Fatalf("concurrent retry error = %v, want upload in progress", err)
		}
	}
	close(objectStore.releaseFirst)
	if err := <-firstErr; err != nil {
		t.Fatalf("original owner upload: %v", err)
	}

	objectKey, _ := records.inserted[constants.FieldSysFile_StorageKey].(string)
	if objectKey == "" {
		t.Fatal("completed metadata has no object key")
	}
	objectStore.mu.Lock()
	defer objectStore.mu.Unlock()
	if objectStore.objects[objectKey] != "same" {
		t.Fatalf("completed object = %q, want original body", objectStore.objects[objectKey])
	}
	if len(objectStore.objects) != 1 || objectStore.puts != 1 {
		t.Fatalf("concurrent retry wrote another attempt: puts %d objects %#v",
			objectStore.puts, objectStore.objects)
	}
	if journal.row.ObjectKey != objectKey || journal.row.State != persistence.FileUploadOperationCompleted {
		t.Fatalf("completed journal = %#v", journal.row)
	}
}

func TestDurableRecordUploadDoesNotOverwriteResultAfterReceiptExpiry(t *testing.T) {
	journal := &durableUploadTestJournal{}
	objectStore := &durableUploadTestObjectStore{}
	records := &captureFileRecordStore{}
	request := durableUploadTestRequest("same")
	service := newDurableUploadTestService(objectStore, records, journal)

	if _, err := service.StoreRecordAttachment(t.Context(), request); err != nil {
		t.Fatal(err)
	}
	if objectStore.puts != 1 {
		t.Fatalf("first upload puts = %d, want 1", objectStore.puts)
	}
	if journal.acceptCalls != 1 {
		t.Fatalf("first upload journal accepts = %d, want 1", journal.acceptCalls)
	}
	committedObjectKey, _ := records.inserted[constants.FieldSysFile_StorageKey].(string)
	if committedObjectKey == "" {
		t.Fatal("first upload committed no storage key")
	}

	// Expiring the operation receipt must not make the committed immutable
	// attempt object overwritable. The _System_File result is replayed before
	// journal admission, so no new owner or object write is created.
	journal.row = persistence.FileUploadOperation{}
	request.Body = strings.NewReader("same")
	if _, err := service.StoreRecordAttachment(t.Context(), request); err != nil {
		t.Fatal(err)
	}
	if objectStore.puts != 1 || journal.acceptCalls != 1 || journal.row.ID != "" {
		t.Fatalf(
			"receipt recovery = puts %d accepts %d journal %#v",
			objectStore.puts, journal.acceptCalls, journal.row,
		)
	}

	journal.row = persistence.FileUploadOperation{}
	request.Body = strings.NewReader("diff")
	if _, err := service.StoreRecordAttachment(t.Context(), request); err == nil {
		t.Fatal("expired receipt allowed committed object overwrite")
	}
	if objectStore.puts != 1 || journal.acceptCalls != 1 ||
		objectStore.objects[committedObjectKey] != "same" {
		t.Fatalf(
			"conflicting receipt recovery mutated state: puts %d accepts %d objects %#v",
			objectStore.puts, journal.acceptCalls, objectStore.objects,
		)
	}
}

func TestDurableRecordUploadReceiptReplayUsesAndHashesRecordStorageAuthority(t *testing.T) {
	journal := &durableUploadTestJournal{}
	currentStore := &durableUploadTestObjectStore{}
	historicalStore := &durableUploadTestObjectStore{}
	records := &captureFileRecordStore{}
	request := durableUploadTestRequest("same")
	service := newDurableUploadTestService(currentStore, records, journal)

	if _, err := service.StoreRecordAttachment(t.Context(), request); err != nil {
		t.Fatal(err)
	}
	objectKey, _ := records.inserted[constants.FieldSysFile_StorageKey].(string)
	if objectKey == "" {
		t.Fatal("committed metadata has no object key")
	}
	historicalStore.objects = map[string]string{objectKey: "same"}
	currentStore.objects[objectKey] = "evil"
	records.inserted[constants.FieldSysFile_StorageProvider] = storage.ProviderFilesystem
	records.inserted[constants.FieldSysFile_StorageBucket] = "historical"
	journal.row = persistence.FileUploadOperation{}
	resolved := false
	service.resolveStore = func(config storage.Config) (storage.ObjectStore, error) {
		resolved = true
		if config.Provider != storage.ProviderFilesystem || config.Bucket != "historical" {
			t.Fatalf("resolved authority = %q/%q", config.Provider, config.Bucket)
		}
		return historicalStore, nil
	}

	request.Body = strings.NewReader("same")
	if _, err := service.StoreRecordAttachment(t.Context(), request); err != nil {
		t.Fatal(err)
	}
	if !resolved {
		t.Fatal("receipt replay bypassed the record-bound storage authority")
	}

	historicalStore.objects[objectKey] = "evil"
	request.Body = strings.NewReader("same")
	if _, err := service.StoreRecordAttachment(t.Context(), request); err == nil {
		t.Fatal("receipt replay accepted a same-size corrupt authoritative blob")
	}
	if journal.acceptCalls != 1 || currentStore.puts != 1 || historicalStore.puts != 0 {
		t.Fatalf(
			"receipt replay mutated state: accepts=%d current puts=%d historical puts=%d",
			journal.acceptCalls,
			currentStore.puts,
			historicalStore.puts,
		)
	}
}

func TestDurableRecordUploadExistingPoisonCannotBeBypassedByMetadataReplay(t *testing.T) {
	journal := &durableUploadTestJournal{}
	objectStore := &durableUploadTestObjectStore{}
	records := &captureFileRecordStore{}
	request := durableUploadTestRequest("same")
	service := newDurableUploadTestService(objectStore, records, journal)
	if _, err := service.StoreRecordAttachment(t.Context(), request); err != nil {
		t.Fatal(err)
	}
	objectKey := journal.row.ObjectKey
	journal.row.State = persistence.FileUploadOperationFailed
	objectStore.objects[objectKey] = "evil" // same size, different committed bytes
	acceptCalls := journal.acceptCalls
	request.Body = strings.NewReader("same")
	_, err := service.StoreRecordAttachment(t.Context(), request)
	var inProgress *uploadInProgressError
	if !errors.As(err, &inProgress) {
		t.Fatalf("poisoned journal retry error = %v, want in-progress repair fence", err)
	}
	if journal.acceptCalls != acceptCalls ||
		journal.row.State != persistence.FileUploadOperationFailed ||
		objectStore.objects[objectKey] != "evil" {
		t.Fatalf(
			"poisoned journal was bypassed: accepts %d/%d state %q object %q",
			journal.acceptCalls, acceptCalls, journal.row.State, objectStore.objects[objectKey],
		)
	}
}

func TestDurableRecordUploadConflictingActiveJournalCannotBeBypassedByMetadataReplay(t *testing.T) {
	journal := &durableUploadTestJournal{}
	objectStore := &durableUploadTestObjectStore{}
	records := &captureFileRecordStore{}
	request := durableUploadTestRequest("same")
	service := newDurableUploadTestService(objectStore, records, journal)
	if _, err := service.StoreRecordAttachment(t.Context(), request); err != nil {
		t.Fatal(err)
	}
	journal.row.State = persistence.FileUploadOperationPrepared
	journal.row.ParentID = "different-parent"
	acceptCalls := journal.acceptCalls
	request.Body = strings.NewReader("same")
	if _, err := service.StoreRecordAttachment(t.Context(), request); err == nil ||
		!strings.Contains(err.Error(), "different content") {
		t.Fatalf("conflicting active journal retry error = %v", err)
	}
	if journal.acceptCalls != acceptCalls ||
		journal.row.ParentID != "different-parent" ||
		objectStore.puts != 1 {
		t.Fatalf(
			"conflicting active journal was bypassed: accepts %d/%d parent %q puts %d",
			journal.acceptCalls, acceptCalls, journal.row.ParentID, objectStore.puts,
		)
	}
}

func TestDurableRecordUploadReconcilesBlobStoredWithoutReupload(t *testing.T) {
	journal := &durableUploadTestJournal{}
	objectStore := &durableUploadTestObjectStore{objects: map[string]string{}}
	records := &captureFileRecordStore{}
	request := durableUploadTestRequest("same")
	parentObject := request.ParentObjectAPIName
	logicalObjectKey := systemFileObjectKey(parentObject, request.ParentID, request.ID, request.FileName)
	headerHash, err := durableUploadHeaderHash(request, parentObject, logicalObjectKey)
	if err != nil {
		t.Fatal(err)
	}
	ownerToken := "owner-restart"
	objectKey := durableUploadAttemptObjectKey(logicalObjectKey, ownerToken)
	contentDigest, err := checksumFileBody(strings.NewReader("same"))
	if err != nil {
		t.Fatal(err)
	}
	journal.row = persistence.FileUploadOperation{
		ID: request.ID, IdempotencyKey: request.ID,
		RequestHash: headerHash + "|" + contentDigest,
		State:       persistence.FileUploadOperationBlobStored,
		OwnerToken:  ownerToken, ObjectKey: objectKey,
		StorageProvider:        storage.ProviderFilesystem,
		StorageBucket:          storage.DefaultFilesystemBucket,
		StorageAuthorityDigest: durableUploadTestAuthorityDigest(t),
		SystemFileID:           request.ID, ParentID: request.ParentID,
		ParentObjectAPIName: parentObject,
	}
	objectStore.objects[objectKey] = "same"

	result, err := newDurableUploadTestService(objectStore, records, journal).
		StoreRecordAttachment(t.Context(), request)
	if err != nil {
		t.Fatal(err)
	}
	if result.ID != request.ID || journal.row.State != persistence.FileUploadOperationCompleted ||
		objectStore.puts != 0 || records.inserted == nil {
		t.Fatalf("blob reconciliation = result %#v state %q puts %d record %#v",
			result, journal.row.State, objectStore.puts, records.inserted)
	}
}

func TestDurableRecordUploadMissingJournaledBlobFailsClosedForScheduledRepair(t *testing.T) {
	journal := &durableUploadTestJournal{}
	objectStore := &durableUploadTestObjectStore{objects: map[string]string{}}
	records := &captureFileRecordStore{}
	request := durableUploadTestRequest("same")
	logicalObjectKey := systemFileObjectKey(
		request.ParentObjectAPIName, request.ParentID, request.ID, request.FileName,
	)
	headerHash, err := durableUploadHeaderHash(request, request.ParentObjectAPIName, logicalObjectKey)
	if err != nil {
		t.Fatal(err)
	}
	ownerToken := "owner-missing"
	objectKey := durableUploadAttemptObjectKey(logicalObjectKey, ownerToken)
	contentDigest, err := checksumFileBody(strings.NewReader("same"))
	if err != nil {
		t.Fatal(err)
	}
	journal.row = persistence.FileUploadOperation{
		ID: request.ID, IdempotencyKey: request.ID,
		RequestHash: headerHash + "|" + contentDigest,
		State:       persistence.FileUploadOperationBlobStored,
		OwnerToken:  ownerToken, ObjectKey: objectKey,
		StorageProvider:        storage.ProviderFilesystem,
		StorageBucket:          storage.DefaultFilesystemBucket,
		StorageAuthorityDigest: durableUploadTestAuthorityDigest(t),
		SystemFileID:           request.ID, ParentID: request.ParentID,
		ParentObjectAPIName: request.ParentObjectAPIName,
	}

	_, err = newDurableUploadTestService(objectStore, records, journal).
		StoreRecordAttachment(t.Context(), request)
	var repairRequired *uploadRepairRequiredError
	if !errors.As(err, &repairRequired) || !repairRequired.RetriableCondition() {
		t.Fatalf("missing BLOB_STORED error = %v, want retriable scheduled-repair error", err)
	}
	if journal.row.State != persistence.FileUploadOperationBlobStored ||
		journal.row.OwnerToken != ownerToken || journal.row.ObjectKey != objectKey ||
		objectStore.puts != 0 || records.inserted != nil {
		t.Fatalf("missing blob mutated foreground authority: row %#v puts %d record %#v",
			journal.row, objectStore.puts, records.inserted)
	}
}

func TestDurableRecordUploadForeignJournaledObjectKeyFailsClosedBeforeIO(t *testing.T) {
	journal := &durableUploadTestJournal{}
	objectStore := &durableUploadTestObjectStore{objects: map[string]string{}}
	records := &captureFileRecordStore{}
	request := durableUploadTestRequest("same")
	logicalObjectKey := systemFileObjectKey(
		request.ParentObjectAPIName,
		request.ParentID,
		request.ID,
		request.FileName,
	)
	headerHash, err := durableUploadHeaderHash(
		request,
		request.ParentObjectAPIName,
		logicalObjectKey,
	)
	if err != nil {
		t.Fatal(err)
	}
	contentDigest, err := checksumFileBody(strings.NewReader("same"))
	if err != nil {
		t.Fatal(err)
	}
	ownerToken := "owner-foreign"
	foreignObjectKey := durableUploadAttemptObjectKey(
		systemFileObjectKey(
			request.ParentObjectAPIName,
			"different-parent",
			"different-file",
			request.FileName,
		),
		ownerToken,
	)
	objectStore.objects[foreignObjectKey] = "same"
	journal.row = persistence.FileUploadOperation{
		ID: request.ID, IdempotencyKey: request.ID,
		RequestHash: headerHash + "|" + contentDigest,
		State:       persistence.FileUploadOperationBlobStored,
		OwnerToken:  ownerToken, ObjectKey: foreignObjectKey,
		StorageProvider:        storage.ProviderFilesystem,
		StorageBucket:          storage.DefaultFilesystemBucket,
		StorageAuthorityDigest: durableUploadTestAuthorityDigest(t),
		SystemFileID:           request.ID, ParentID: request.ParentID,
		ParentObjectAPIName: request.ParentObjectAPIName,
	}

	_, err = newDurableUploadTestService(objectStore, records, journal).
		StoreRecordAttachment(t.Context(), request)
	var repairRequired *uploadRepairRequiredError
	if !errors.As(err, &repairRequired) || !repairRequired.RetriableCondition() {
		t.Fatalf("foreign-key retry error = %v, want scheduled-repair fence", err)
	}
	if journal.row.State != persistence.FileUploadOperationBlobStored ||
		journal.row.ObjectKey != foreignObjectKey ||
		objectStore.gets != 0 || objectStore.puts != 0 || objectStore.deletes != 0 ||
		records.inserted != nil {
		t.Fatalf(
			"foreign-key retry crossed fail-closed boundary: row %#v gets=%d puts=%d deletes=%d record=%#v",
			journal.row,
			objectStore.gets,
			objectStore.puts,
			objectStore.deletes,
			records.inserted,
		)
	}
	if objectStore.objects[foreignObjectKey] != "same" {
		t.Fatal("foreign object was mutated")
	}
}

func TestDurableRecordUploadRestartWithChangedStorageMappingFailsBeforeIO(t *testing.T) {
	t.Setenv("FILE_STORAGE_PROVIDER", storage.ProviderFilesystem)
	t.Setenv("FILE_STORAGE_ROOT", t.TempDir())
	journal := &durableUploadTestJournal{}
	objectStore := &durableUploadTestObjectStore{objects: map[string]string{}}
	records := &captureFileRecordStore{}
	request := durableUploadTestRequest("same")
	if _, err := newDurableUploadTestService(objectStore, records, journal).
		StoreRecordAttachment(t.Context(), request); err != nil {
		t.Fatal(err)
	}
	if objectStore.puts != 1 {
		t.Fatalf("initial puts = %d", objectStore.puts)
	}

	t.Setenv("FILE_STORAGE_ROOT", t.TempDir())
	request.Body = strings.NewReader("same")
	if _, err := newDurableUploadTestService(objectStore, records, journal).
		StoreRecordAttachment(t.Context(), request); err == nil ||
		!strings.Contains(err.Error(), "different content") {
		t.Fatalf("mapping-changed retry error = %v", err)
	}
	if objectStore.puts != 1 || objectStore.deletes != 0 {
		t.Fatalf(
			"mapping-changed retry performed object I/O: puts=%d deletes=%d",
			objectStore.puts,
			objectStore.deletes,
		)
	}
}

func TestMarkDurableUploadCompletedAcceptsOnlyExactConcurrentCompletion(t *testing.T) {
	row := persistence.FileUploadOperation{
		ID: "file-1", IdempotencyKey: "file-1",
		State:      persistence.FileUploadOperationBlobStored,
		OwnerToken: "owner-1", ObjectKey: "attempt-1",
	}
	completeHash := "sha256:header|sha256:content"
	exact := row
	exact.State = persistence.FileUploadOperationCompleted
	exact.RequestHash = completeHash
	for _, test := range []struct {
		name      string
		reloaded  persistence.FileUploadOperation
		wantError bool
	}{
		{name: "exact concurrent completion", reloaded: exact},
		{name: "different owner", reloaded: func() persistence.FileUploadOperation {
			value := exact
			value.OwnerToken = "owner-2"
			return value
		}(), wantError: true},
		{name: "different object key", reloaded: func() persistence.FileUploadOperation {
			value := exact
			value.ObjectKey = "attempt-2"
			return value
		}(), wantError: true},
		{name: "different digest", reloaded: func() persistence.FileUploadOperation {
			value := exact
			value.RequestHash = "sha256:header|sha256:other"
			return value
		}(), wantError: true},
	} {
		t.Run(test.name, func(t *testing.T) {
			journal := &completionRaceUploadJournal{
				durableUploadTestJournal: &durableUploadTestJournal{},
				reloaded:                 test.reloaded,
			}
			service := &ContractService{uploadOperations: journal}
			err := service.markDurableUploadCompleted(t.Context(), row, completeHash)
			if test.wantError && !errors.Is(err, persistence.ErrFileUploadOperationLostOwnership) {
				t.Fatalf("markDurableUploadCompleted() error = %v, want lost ownership", err)
			}
			if !test.wantError && err != nil {
				t.Fatalf("markDurableUploadCompleted() error = %v", err)
			}
		})
	}
}

type denyFileCreatePermissionChecker struct{}

func (denyFileCreatePermissionChecker) CheckPermissionOrErrorWithUser(
	context.Context,
	string,
	string,
	*models.UserSession,
) error {
	return errors.New("denied")
}

func TestDurableRecordUploadAuthorizesBeforeJournalDisclosure(t *testing.T) {
	journal := &durableUploadTestJournal{}
	service := NewContractService(
		&durableUploadTestObjectStore{}, storage.ProviderFilesystem, "", "",
		&captureFileRecordStore{}, denyFileCreatePermissionChecker{}, nil,
		WithDurableUploadJournal(journal),
	)
	if _, err := service.StoreRecordAttachment(t.Context(), durableUploadTestRequest("same")); err == nil {
		t.Fatal("unauthorized upload succeeded")
	}
	if journal.acceptCalls != 0 {
		t.Fatalf("unauthorized upload read journal %d times", journal.acceptCalls)
	}
}

func TestDurableUploadContentDigestRejectsNonHexJournalData(t *testing.T) {
	headerHash := "sha256:" + strings.Repeat("a", 64)
	if _, ok := durableUploadContentDigest(
		headerHash+"|sha256:"+strings.Repeat("z", 64), headerHash,
	); ok {
		t.Fatal("journal accepted a non-hex content digest")
	}
}

func TestDurableUploadAttemptDeadlinePrecedesCleanupEligibility(t *testing.T) {
	now := time.Now().UTC()
	parentDeadline := now.Add(5 * time.Minute)
	parent, cancelParent := context.WithDeadline(context.Background(), parentDeadline)
	defer cancelParent()
	leaseUntil := durableUploadLeaseUntil(parent, now)
	if !leaseUntil.Equal(parentDeadline.Add(durableUploadShutdownGrace)) {
		t.Fatalf("request-bound lease = %v, want %v",
			leaseUntil, parentDeadline.Add(durableUploadShutdownGrace))
	}
	attemptCtx, cancelAttempt := durableUploadAttemptContext(parent, leaseUntil)
	defer cancelAttempt()
	attemptDeadline, ok := attemptCtx.Deadline()
	if !ok || !attemptDeadline.Equal(parentDeadline) {
		t.Fatalf("request-bound attempt deadline = %v present %t, want %v",
			attemptDeadline, ok, parentDeadline)
	}

	backgroundLease := durableUploadLeaseUntil(context.Background(), now)
	backgroundAttempt, cancelBackground := durableUploadAttemptContext(
		context.Background(), backgroundLease,
	)
	defer cancelBackground()
	backgroundDeadline, ok := backgroundAttempt.Deadline()
	wantBackgroundDeadline := now.Add(15 * time.Minute)
	if !ok || !backgroundDeadline.Equal(wantBackgroundDeadline) {
		t.Fatalf("fallback attempt deadline = %v present %t, want %v",
			backgroundDeadline, ok, wantBackgroundDeadline)
	}
	if !backgroundLease.Equal(backgroundDeadline.Add(durableUploadShutdownGrace)) {
		t.Fatalf("fallback cleanup eligibility = %v, want %v",
			backgroundLease, backgroundDeadline.Add(durableUploadShutdownGrace))
	}
}

func newDurableUploadTestService(
	objectStore storage.ObjectStore,
	records fileRecordStore,
	journal UploadOperationJournal,
) *ContractService {
	return NewContractService(
		objectStore, storage.ProviderFilesystem, "", "",
		records, allowFilePermissionChecker{}, nil,
		WithDurableUploadJournal(journal),
	)
}

func durableUploadTestAuthorityDigest(t *testing.T) string {
	t.Helper()
	config := storage.LoadConfigFromEnv()
	config.Provider = storage.ProviderFilesystem
	config.Bucket = storage.DefaultFilesystemBucket
	digest, err := storage.ConfigAuthorityDigest(config)
	if err != nil {
		t.Fatal(err)
	}
	return digest
}

func durableUploadTestRequest(body string) contract.FileUploadRequest {
	return contract.FileUploadRequest{
		ID: "file-durable-id", FileName: "contract.txt", ContentType: "text/plain",
		Body: strings.NewReader(body), ContentLength: int64(len(body)),
		ParentID: "800000000000001", ParentObjectAPIName: "Contract",
		AttachmentKind: defaultRecordAttachmentKind,
		User:           &models.UserSession{ID: "005000000000001", ProfileID: constants.ProfileSystemAdmin},
	}
}