package files

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"io"
	"path"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

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

func TestRuntimeUploadLifecycleTiDBCompletesCleansAndPoisonsWithoutFallback(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	objectStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	ownership := activateRuntimeUploadLifecycleOwner(t, db, "runtime-files-1", 1)
	lifecycle, err := NewRuntimeUploadLifecycle(db, objectStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	if err := lifecycle.CheckReady(t.Context()); err != nil {
		t.Fatal(err)
	}

	completed := seedRuntimeUploadLifecycleOperation(t, db, objectStore, "complete", true, true)
	orphan := seedRuntimeUploadLifecycleOperation(t, db, objectStore, "orphan", false, true)
	poisoned := seedRuntimeUploadLifecycleOperation(t, db, objectStore, "poison", true, false)

	processed, err := lifecycle.ProcessExpired(t.Context(), 10, time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	if processed != 3 {
		t.Fatalf("processed = %d, want 3", processed)
	}
	var state, errorCode string
	if err := db.QueryRow(`SELECT State, COALESCE(ErrorCode, '')
FROM _System_FileUploadOperation WHERE Id = ?`, completed.ID).Scan(&state, &errorCode); err != nil {
		t.Fatal(err)
	}
	if state != persistence.FileUploadOperationCompleted || errorCode != "" {
		t.Fatalf("completed row = state %q error %q", state, errorCode)
	}
	var orphanCount int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_FileUploadOperation WHERE Id = ?`, orphan.ID).Scan(&orphanCount); err != nil {
		t.Fatal(err)
	}
	if orphanCount != 0 {
		t.Fatalf("orphan journal count = %d", orphanCount)
	}
	if _, err := objectStore.HeadObject(t.Context(), orphan.ObjectKey); !errors.Is(err, storage.ErrObjectNotFound) {
		t.Fatalf("orphan object remains: %v", err)
	}
	var poisonMessage string
	if err := db.QueryRow(`SELECT State, ErrorCode, ErrorMessage
FROM _System_FileUploadOperation WHERE Id = ?`, poisoned.ID).Scan(&state, &errorCode, &poisonMessage); err != nil {
		t.Fatal(err)
	}
	if state != persistence.FileUploadOperationFailed || errorCode != fileUploadRepairPoisoned ||
		poisonMessage != "committed_blob_missing" {
		t.Fatalf("poison row = state %q code %q message %q", state, errorCode, poisonMessage)
	}

	observer, err := NewRuntimeUploadLifecycleObserver(db)
	if err != nil {
		t.Fatal(err)
	}
	observed, err := observer.Observe(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	if observed.QueueDepth != 0 || observed.PoisonDepth != 1 {
		t.Fatalf("observed = %#v", observed)
	}
}

func TestRuntimeUploadLifecycleTiDBUsesAuthoritativeMetadataStorageLocation(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	defaultStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	historicalStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	ownership := activateRuntimeUploadLifecycleOwner(t, db, "runtime-files-storage-authority", 1)
	lifecycle, err := NewRuntimeUploadLifecycle(db, defaultStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	lifecycle.defaultStorageConfig.Provider = storage.ProviderFilesystem
	lifecycle.defaultStorageConfig.Bucket = "current"
	lifecycle.resolveStore = func(config storage.Config) (storage.ObjectStore, error) {
		if config.Provider != storage.ProviderFilesystem || config.Bucket != "historical" {
			t.Fatalf("resolved storage authority = %q/%q", config.Provider, config.Bucket)
		}
		return historicalStore, nil
	}
	operation := seedRuntimeUploadLifecycleOperation(
		t, db, historicalStore, "historical-store", true, true,
	)
	historicalDigest := runtimeUploadStorageAuthorityDigest(
		t, lifecycle.defaultStorageConfig, storage.ProviderFilesystem, "historical",
	)
	if _, err := db.Exec(`UPDATE _System_File
SET StorageProvider = ?, StorageBucket = ?
WHERE Id = ?`, storage.ProviderFilesystem, "historical", operation.SystemFileID); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET StorageProvider = ?, StorageBucket = ?, StorageAuthorityDigest = ?
WHERE Id = ?`, storage.ProviderFilesystem, "historical", historicalDigest, operation.ID); err != nil {
		t.Fatal(err)
	}

	processed, err := lifecycle.ProcessExpired(t.Context(), 1, time.Minute)
	if err != nil || processed != 1 {
		t.Fatalf("process = %d, %v", processed, err)
	}
	assertRuntimeUploadJournalState(
		t, db, operation.ID, persistence.FileUploadOperationCompleted, "",
	)
	if _, err := defaultStore.HeadObject(t.Context(), operation.ObjectKey); !errors.Is(
		err, storage.ErrObjectNotFound,
	) {
		t.Fatalf("default store unexpectedly owned the blob: %v", err)
	}
}

func TestRuntimeUploadLifecycleTiDBUsesJournalStorageAuthorityWhenMetadataIsMissing(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	defaultStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	historicalStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	ownership := activateRuntimeUploadLifecycleOwner(t, db, "runtime-files-orphan-authority", 1)
	lifecycle, err := NewRuntimeUploadLifecycle(db, defaultStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	lifecycle.defaultStorageConfig.Provider = storage.ProviderFilesystem
	lifecycle.defaultStorageConfig.Bucket = "current"
	lifecycle.resolveStore = func(config storage.Config) (storage.ObjectStore, error) {
		if config.Provider != storage.ProviderFilesystem || config.Bucket != "historical" {
			t.Fatalf("resolved storage authority = %q/%q", config.Provider, config.Bucket)
		}
		return historicalStore, nil
	}
	operation := seedRuntimeUploadLifecycleOperation(
		t, db, historicalStore, "historical-orphan", false, true,
	)
	historicalDigest := runtimeUploadStorageAuthorityDigest(
		t, lifecycle.defaultStorageConfig, storage.ProviderFilesystem, "historical",
	)
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET StorageProvider = ?, StorageBucket = ?, StorageAuthorityDigest = ?
WHERE Id = ?`, storage.ProviderFilesystem, "historical", historicalDigest, operation.ID); err != nil {
		t.Fatal(err)
	}
	decoy := "must-not-delete"
	if _, err := defaultStore.PutObject(
		t.Context(), operation.ObjectKey, strings.NewReader(decoy),
		storage.PutObjectOptions{ContentLength: int64Pointer(int64(len(decoy)))},
	); err != nil {
		t.Fatal(err)
	}

	processed, err := lifecycle.ProcessExpired(t.Context(), 1, time.Minute)
	if err != nil || processed != 1 {
		t.Fatalf("process = %d, %v", processed, err)
	}
	assertRuntimeUploadJournalAbsent(t, db, operation.ID)
	if _, err := historicalStore.HeadObject(t.Context(), operation.ObjectKey); !errors.Is(
		err, storage.ErrObjectNotFound,
	) {
		t.Fatalf("historical orphan remains: %v", err)
	}
	reader, _, err := defaultStore.GetObject(t.Context(), operation.ObjectKey)
	if err != nil {
		t.Fatal(err)
	}
	body, readErr := io.ReadAll(reader)
	closeErr := reader.Close()
	if readErr != nil || closeErr != nil || string(body) != decoy {
		t.Fatalf("default-store decoy = %q read=%v close=%v", body, readErr, closeErr)
	}
}

func TestRuntimeUploadLifecycleTiDBStorageMappingChangeFailsClosedBeforeDelete(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)

	oldRoot := t.TempDir()
	newRoot := t.TempDir()
	t.Setenv("FILE_STORAGE_PROVIDER", storage.ProviderFilesystem)
	t.Setenv("FILE_STORAGE_BUCKET", storage.DefaultFilesystemBucket)
	t.Setenv("FILE_STORAGE_ROOT", oldRoot)
	oldStore, err := storage.NewStoreFromEnv()
	if err != nil {
		t.Fatal(err)
	}
	operation := seedRuntimeUploadPreparedOperation(
		t, db, "mapping-change", storage.DefaultFilesystemBucket,
	)
	putRuntimeUploadObject(t, oldStore, operation.ObjectKey, "historical")

	t.Setenv("FILE_STORAGE_ROOT", newRoot)
	newStore, err := storage.NewStoreFromEnv()
	if err != nil {
		t.Fatal(err)
	}
	putRuntimeUploadObject(t, newStore, operation.ObjectKey, "current-decoy")
	ownership := activateRuntimeUploadLifecycleOwner(
		t, db, "runtime-files-mapping-change", 1,
	)
	lifecycle, err := NewRuntimeUploadLifecycle(db, newStore, ownership)
	if err != nil {
		t.Fatal(err)
	}

	processed, err := lifecycle.ProcessExpired(t.Context(), 1, time.Minute)
	if err != nil || processed != 1 {
		t.Fatalf("process = %d, %v", processed, err)
	}
	var state, errorCode, errorMessage string
	if err := db.QueryRow(`SELECT State, ErrorCode, ErrorMessage
FROM _System_FileUploadOperation WHERE Id = ?`, operation.ID).Scan(
		&state, &errorCode, &errorMessage,
	); err != nil {
		t.Fatal(err)
	}
	if state != persistence.FileUploadOperationFailed ||
		errorCode != fileUploadPreparedRepairPoisoned ||
		errorMessage != "storage_authority_mismatch" {
		t.Fatalf(
			"mapping-change disposition = %q/%q/%q",
			state, errorCode, errorMessage,
		)
	}
	assertRuntimeUploadObjectBody(t, oldStore, operation.ObjectKey, "historical")
	assertRuntimeUploadObjectBody(t, newStore, operation.ObjectKey, "current-decoy")
}

func TestRuntimeUploadLifecycleTiDBMissingStorageAuthorityPoisonsWithoutDeleting(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	objectStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	ownership := activateRuntimeUploadLifecycleOwner(t, db, "runtime-files-missing-authority", 1)
	lifecycle, err := NewRuntimeUploadLifecycle(db, objectStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	operation := seedRuntimeUploadLifecycleOperation(
		t, db, objectStore, "missing-authority", false, true,
	)
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET StorageProvider = NULL, StorageBucket = NULL
WHERE Id = ?`, operation.ID); err != nil {
		t.Fatal(err)
	}

	processed, err := lifecycle.ProcessExpired(t.Context(), 1, time.Minute)
	if err != nil || processed != 1 {
		t.Fatalf("process = %d, %v", processed, err)
	}
	var state, errorCode, errorMessage string
	if err := db.QueryRow(`SELECT State, ErrorCode, ErrorMessage
FROM _System_FileUploadOperation WHERE Id = ?`, operation.ID).Scan(
		&state, &errorCode, &errorMessage,
	); err != nil {
		t.Fatal(err)
	}
	if state != persistence.FileUploadOperationFailed ||
		errorCode != fileUploadRepairPoisoned ||
		errorMessage != "storage_authority_missing" {
		t.Fatalf("missing authority disposition = %q/%q/%q", state, errorCode, errorMessage)
	}
	if _, err := objectStore.HeadObject(t.Context(), operation.ObjectKey); err != nil {
		t.Fatalf("blob without authority was deleted: %v", err)
	}
}

func TestRuntimeUploadLifecycleTiDBInvalidMatchingStorageAuthorityPoisons(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	objectStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	ownership := activateRuntimeUploadLifecycleOwner(
		t,
		db,
		"runtime-files-invalid-authority",
		1,
	)
	lifecycle, err := NewRuntimeUploadLifecycle(db, objectStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	lifecycle.resolveStore = func(storage.Config) (storage.ObjectStore, error) {
		t.Fatal("invalid persisted authority reached store resolution")
		return nil, nil
	}

	tests := []struct {
		name, provider, bucket string
	}{
		{name: "blank", provider: "", bucket: ""},
		{name: "unsupported", provider: "unsupported", bucket: "historical"},
	}
	operations := make(map[string]persistence.FileUploadOperation, len(tests))
	for _, test := range tests {
		operation := seedRuntimeUploadLifecycleOperation(
			t,
			db,
			objectStore,
			"invalid-authority-"+test.name,
			true,
			true,
		)
		if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET StorageProvider = ?, StorageBucket = ?
WHERE Id = ?`, test.provider, test.bucket, operation.ID); err != nil {
			t.Fatal(err)
		}
		if _, err := db.Exec(`UPDATE _System_File
SET StorageProvider = ?, StorageBucket = ?
WHERE Id = ?`, test.provider, test.bucket, operation.SystemFileID); err != nil {
			t.Fatal(err)
		}
		operations[test.name] = operation
	}

	processed, err := lifecycle.ProcessExpired(t.Context(), len(tests), time.Minute)
	if err != nil || processed != int64(len(tests)) {
		t.Fatalf("process = %d, %v", processed, err)
	}
	for _, test := range tests {
		var state, errorCode, errorMessage string
		var retryCount int
		operation := operations[test.name]
		if err := db.QueryRow(`SELECT State, ErrorCode, ErrorMessage, RetryCount
FROM _System_FileUploadOperation WHERE Id = ?`, operation.ID).Scan(
			&state,
			&errorCode,
			&errorMessage,
			&retryCount,
		); err != nil {
			t.Fatal(err)
		}
		if state != persistence.FileUploadOperationFailed ||
			errorCode != fileUploadRepairPoisoned ||
			errorMessage != "storage_authority_missing" ||
			retryCount != 1 {
			t.Fatalf(
				"%s disposition = %q/%q/%q retry=%d",
				test.name,
				state,
				errorCode,
				errorMessage,
				retryCount,
			)
		}
		if _, err := objectStore.HeadObject(
			t.Context(),
			operation.ObjectKey,
		); err != nil {
			t.Fatalf("%s authoritative blob was touched: %v", test.name, err)
		}
	}
}

func TestRuntimeUploadLifecycleTiDBValidAuthorityResolverFailureRemainsRetryable(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	defaultStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	historicalStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	ownership := activateRuntimeUploadLifecycleOwner(
		t,
		db,
		"runtime-files-transient-authority",
		1,
	)
	lifecycle, err := NewRuntimeUploadLifecycle(db, defaultStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	operation := seedRuntimeUploadLifecycleOperation(
		t,
		db,
		historicalStore,
		"transient-authority",
		true,
		true,
	)
	historicalDigest := runtimeUploadStorageAuthorityDigest(
		t, lifecycle.defaultStorageConfig, storage.ProviderFilesystem, "historical",
	)
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET StorageProvider = ?, StorageBucket = ?, StorageAuthorityDigest = ?
WHERE Id = ?`, storage.ProviderFilesystem, "historical", historicalDigest, operation.ID); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_File
SET StorageProvider = ?, StorageBucket = ?
WHERE Id = ?`, storage.ProviderFilesystem, "historical", operation.SystemFileID); err != nil {
		t.Fatal(err)
	}
	sentinel := errors.New("transient store construction")
	lifecycle.resolveStore = func(storage.Config) (storage.ObjectStore, error) {
		return nil, sentinel
	}
	processed, err := lifecycle.ProcessExpired(t.Context(), 1, time.Minute)
	if processed != 0 || !errors.Is(err, sentinel) {
		t.Fatalf("transient process = %d, %v", processed, err)
	}
	assertRuntimeUploadJournalState(
		t,
		db,
		operation.ID,
		persistence.FileUploadOperationFailed,
		fileUploadCleanupClaimed,
	)
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET ExpiresAt = DATE_SUB(UTC_TIMESTAMP(6), INTERVAL 1 SECOND)
WHERE Id = ?`, operation.ID); err != nil {
		t.Fatal(err)
	}
	lifecycle.resolveStore = func(config storage.Config) (storage.ObjectStore, error) {
		if config.Provider != storage.ProviderFilesystem || config.Bucket != "historical" {
			t.Fatalf("resolved authority = %q/%q", config.Provider, config.Bucket)
		}
		return historicalStore, nil
	}
	processed, err = lifecycle.ProcessExpired(t.Context(), 1, time.Minute)
	if err != nil || processed != 1 {
		t.Fatalf("restart process = %d, %v", processed, err)
	}
	assertRuntimeUploadJournalState(
		t,
		db,
		operation.ID,
		persistence.FileUploadOperationCompleted,
		"",
	)
}

func TestRuntimeUploadLifecycleTiDBPreparedCleanupUsesTwoFencedPasses(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	defaultStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	historicalStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	ownership := activateRuntimeUploadLifecycleOwner(
		t,
		db,
		"runtime-files-prepared-two-pass",
		1,
	)
	lifecycle, err := NewRuntimeUploadLifecycle(db, defaultStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	lifecycle.defaultStorageConfig.Provider = storage.ProviderFilesystem
	lifecycle.defaultStorageConfig.Bucket = "current"
	lifecycle.resolveStore = func(config storage.Config) (storage.ObjectStore, error) {
		if config.Provider != storage.ProviderFilesystem || config.Bucket != "historical" {
			t.Fatalf("resolved authority = %q/%q", config.Provider, config.Bucket)
		}
		return historicalStore, nil
	}
	operation := seedRuntimeUploadPreparedOperation(
		t,
		db,
		"prepared-two-pass",
		"historical",
	)
	putRuntimeUploadObject(t, historicalStore, operation.ObjectKey, "first-attempt")
	putRuntimeUploadObject(t, defaultStore, operation.ObjectKey, "current-store-decoy")

	processed, err := lifecycle.ProcessExpired(t.Context(), 1, time.Minute)
	if err != nil || processed != 1 {
		t.Fatalf("first prepared pass = %d, %v", processed, err)
	}
	assertRuntimeUploadJournalState(
		t,
		db,
		operation.ID,
		persistence.FileUploadOperationFailed,
		fileUploadPreparedCleanupQuiet,
	)
	if _, err := historicalStore.HeadObject(
		t.Context(),
		operation.ObjectKey,
	); !errors.Is(err, storage.ErrObjectNotFound) {
		t.Fatalf("first-pass historical object remains: %v", err)
	}
	assertRuntimeUploadObjectBody(
		t,
		defaultStore,
		operation.ObjectKey,
		"current-store-decoy",
	)

	// Simulate an object-store commit that becomes visible after the expired
	// request lost its journal authority and after the first delete completed.
	putRuntimeUploadObject(t, historicalStore, operation.ObjectKey, "late-put")
	if processed, err := lifecycle.ProcessExpired(
		t.Context(),
		1,
		time.Minute,
	); err != nil || processed != 0 {
		t.Fatalf("quiet-period process = %d, %v", processed, err)
	}
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET ExpiresAt = DATE_SUB(UTC_TIMESTAMP(6), INTERVAL 1 SECOND)
WHERE Id = ?`, operation.ID); err != nil {
		t.Fatal(err)
	}
	processed, err = lifecycle.ProcessExpired(t.Context(), 1, time.Minute)
	if err != nil || processed != 1 {
		t.Fatalf("final prepared pass = %d, %v", processed, err)
	}
	assertRuntimeUploadJournalAbsent(t, db, operation.ID)
	if _, err := historicalStore.HeadObject(
		t.Context(),
		operation.ObjectKey,
	); !errors.Is(err, storage.ErrObjectNotFound) {
		t.Fatalf("late object remains after final pass: %v", err)
	}
	assertRuntimeUploadObjectBody(
		t,
		defaultStore,
		operation.ObjectKey,
		"current-store-decoy",
	)

	retry := persistence.FileUploadOperationAcceptance{
		ID: operation.ID, IdempotencyKey: operation.IdempotencyKey,
		RequestHash: operation.RequestHash, OwnerToken: "replacement-owner",
		ObjectKey:              operation.ObjectKey + "-replacement",
		StorageProvider:        operation.StorageProvider,
		StorageBucket:          operation.StorageBucket,
		StorageAuthorityDigest: operation.StorageAuthorityDigest,
		SystemFileID:           operation.SystemFileID,
		ParentID:               operation.ParentID, ParentObjectAPIName: operation.ParentObjectAPIName,
		Now: time.Now().UTC(), LeaseUntil: time.Now().UTC().Add(time.Minute),
	}
	if _, owner, err := persistence.NewFileUploadOperationRepository(db).Accept(
		t.Context(),
		retry,
	); err != nil || !owner {
		t.Fatalf("fresh owner after terminal cleanup = owner %t error %v", owner, err)
	}
}

func TestRuntimeUploadLifecycleTiDBPreparedForeignObjectAuthorityPoisonsWithoutDelete(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	objectStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	ownership := activateRuntimeUploadLifecycleOwner(
		t,
		db,
		"runtime-files-prepared-foreign",
		1,
	)
	lifecycle, err := NewRuntimeUploadLifecycle(db, objectStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	operation := seedRuntimeUploadPreparedOperation(
		t,
		db,
		"prepared-foreign",
		storage.DefaultFilesystemBucket,
	)
	foreignObjectKey := path.Join(
		"system-file",
		"Widget",
		"foreign-record",
		"foreign-file",
		".attempts",
		operation.OwnerToken,
		"foreign.txt",
	)
	putRuntimeUploadObject(t, objectStore, foreignObjectKey, "foreign")
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET ObjectKey = ?
WHERE Id = ?`, foreignObjectKey, operation.ID); err != nil {
		t.Fatal(err)
	}

	processed, err := lifecycle.ProcessExpired(t.Context(), 1, time.Minute)
	if err != nil || processed != 1 {
		t.Fatalf("foreign-authority process = %d, %v", processed, err)
	}
	var state, errorCode, errorMessage string
	if err := db.QueryRow(`SELECT State, ErrorCode, ErrorMessage
FROM _System_FileUploadOperation WHERE Id = ?`, operation.ID).Scan(
		&state,
		&errorCode,
		&errorMessage,
	); err != nil {
		t.Fatal(err)
	}
	if state != persistence.FileUploadOperationFailed ||
		errorCode != fileUploadPreparedRepairPoisoned ||
		errorMessage != "journal_identity_invalid" {
		t.Fatalf(
			"foreign authority disposition = %q/%q/%q",
			state,
			errorCode,
			errorMessage,
		)
	}
	assertRuntimeUploadObjectBody(
		t,
		objectStore,
		foreignObjectKey,
		"foreign",
	)
}

func TestRuntimeUploadLifecycleTiDBPreparedCancellationRestartsFirstPass(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	baseStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	blockingStore := newBlockingDeleteObjectStore(baseStore)
	ownership := activateRuntimeUploadLifecycleOwner(
		t,
		db,
		"runtime-files-prepared-cancel",
		1,
	)
	lifecycle, err := NewRuntimeUploadLifecycle(db, blockingStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	operation := seedRuntimeUploadPreparedOperation(
		t,
		db,
		"prepared-cancel",
		storage.DefaultFilesystemBucket,
	)
	putRuntimeUploadObject(t, baseStore, operation.ObjectKey, "prepared")

	ctx, cancel := context.WithCancel(context.Background())
	type processResult struct {
		processed int64
		err       error
	}
	resultChannel := make(chan processResult, 1)
	go func() {
		processed, err := lifecycle.ProcessExpired(ctx, 1, time.Minute)
		resultChannel <- processResult{processed: processed, err: err}
	}()
	select {
	case <-blockingStore.started:
	case <-time.After(5 * time.Second):
		t.Fatal("prepared cleanup did not reach cancellable object deletion")
	}
	cancel()
	result := <-resultChannel
	if result.processed != 0 || !errors.Is(result.err, context.Canceled) {
		t.Fatalf("cancelled first pass = %d, %v", result.processed, result.err)
	}
	assertRuntimeUploadJournalState(
		t,
		db,
		operation.ID,
		persistence.FileUploadOperationFailed,
		fileUploadPreparedCleanupClaimed,
	)

	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET ExpiresAt = DATE_SUB(UTC_TIMESTAMP(6), INTERVAL 1 SECOND)
WHERE Id = ?`, operation.ID); err != nil {
		t.Fatal(err)
	}
	restarted, err := NewRuntimeUploadLifecycle(db, baseStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	processed, err := restarted.ProcessExpired(t.Context(), 1, time.Minute)
	if err != nil || processed != 1 {
		t.Fatalf("restarted first pass = %d, %v", processed, err)
	}
	assertRuntimeUploadJournalState(
		t,
		db,
		operation.ID,
		persistence.FileUploadOperationFailed,
		fileUploadPreparedCleanupQuiet,
	)
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET ExpiresAt = DATE_SUB(UTC_TIMESTAMP(6), INTERVAL 1 SECOND)
WHERE Id = ?`, operation.ID); err != nil {
		t.Fatal(err)
	}
	processed, err = restarted.ProcessExpired(t.Context(), 1, time.Minute)
	if err != nil || processed != 1 {
		t.Fatalf("restarted final pass = %d, %v", processed, err)
	}
	assertRuntimeUploadJournalAbsent(t, db, operation.ID)
}

func TestRuntimeUploadLifecycleTiDBPreparedOwnershipTransferRestartsFirstPass(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	baseStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	blockingStore := newBlockingDeleteObjectStore(baseStore)
	oldOwnership := activateRuntimeUploadLifecycleOwner(
		t,
		db,
		"runtime-files-prepared-old",
		1,
	)
	oldLifecycle, err := NewRuntimeUploadLifecycle(db, blockingStore, oldOwnership)
	if err != nil {
		t.Fatal(err)
	}
	operation := seedRuntimeUploadPreparedOperation(
		t,
		db,
		"prepared-transfer",
		storage.DefaultFilesystemBucket,
	)
	putRuntimeUploadObject(t, baseStore, operation.ObjectKey, "prepared")

	type processResult struct {
		processed int64
		err       error
	}
	resultChannel := make(chan processResult, 1)
	go func() {
		processed, err := oldLifecycle.ProcessExpired(
			context.Background(),
			1,
			time.Minute,
		)
		resultChannel <- processResult{processed: processed, err: err}
	}()
	select {
	case <-blockingStore.started:
	case <-time.After(5 * time.Second):
		t.Fatal("old owner did not reach prepared object deletion")
	}
	newOwnership := persistence.RuntimeWorkOwnership{
		WorkFamily:          RuntimeFileUploadLifecycleWorkFamily,
		DeploymentID:        "runtime-files-prepared-new",
		WorkerReleaseDigest: "sha256:" + strings.Repeat("d", 64),
		OwnershipEpoch:      2,
	}
	if _, err := db.Exec(`UPDATE _System_RuntimeWorkOwnership
SET DeploymentId = ?, WorkerReleaseDigest = ?, OwnershipEpoch = ?, OperationId = ?
WHERE WorkFamily = ? AND DeploymentId = ? AND OwnershipEpoch = 1`,
		newOwnership.DeploymentID,
		newOwnership.WorkerReleaseDigest,
		newOwnership.EpochDecimal(),
		"transfer-prepared-owner",
		newOwnership.WorkFamily,
		oldOwnership.DeploymentID,
	); err != nil {
		t.Fatal(err)
	}
	close(blockingStore.release)
	result := <-resultChannel
	if result.processed != 0 ||
		!errors.Is(result.err, persistence.ErrRuntimeWorkOwnershipLost) {
		t.Fatalf("old-owner first pass = %d, %v", result.processed, result.err)
	}
	assertRuntimeUploadJournalState(
		t,
		db,
		operation.ID,
		persistence.FileUploadOperationFailed,
		fileUploadPreparedCleanupClaimed,
	)

	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET ExpiresAt = DATE_SUB(UTC_TIMESTAMP(6), INTERVAL 1 SECOND)
WHERE Id = ?`, operation.ID); err != nil {
		t.Fatal(err)
	}
	newLifecycle, err := NewRuntimeUploadLifecycle(db, baseStore, newOwnership)
	if err != nil {
		t.Fatal(err)
	}
	if processed, err := newLifecycle.ProcessExpired(
		t.Context(),
		1,
		time.Minute,
	); err != nil || processed != 1 {
		t.Fatalf("new-owner first pass = %d, %v", processed, err)
	}
	assertRuntimeUploadJournalState(
		t,
		db,
		operation.ID,
		persistence.FileUploadOperationFailed,
		fileUploadPreparedCleanupQuiet,
	)
}

func TestRuntimeUploadLifecycleTiDBPreparedPoisonRepairReplaysSamePhase(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	objectStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	ownership := activateRuntimeUploadLifecycleOwner(
		t,
		db,
		"runtime-files-prepared-repair",
		1,
	)
	lifecycle, err := NewRuntimeUploadLifecycle(db, objectStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	operation := seedRuntimeUploadPreparedOperation(
		t,
		db,
		"prepared-repair",
		storage.DefaultFilesystemBucket,
	)
	putRuntimeUploadObject(t, objectStore, operation.ObjectKey, "prepared")
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET StorageProvider = NULL, StorageBucket = NULL
WHERE Id = ?`, operation.ID); err != nil {
		t.Fatal(err)
	}
	if processed, err := lifecycle.ProcessExpired(
		t.Context(),
		1,
		time.Minute,
	); err != nil || processed != 1 {
		t.Fatalf("prepared poison pass = %d, %v", processed, err)
	}
	var retryCount int
	var lastError string
	if err := db.QueryRow(`SELECT RetryCount, ErrorMessage
FROM _System_FileUploadOperation
WHERE Id = ? AND State = 'FAILED' AND ErrorCode = ?`,
		operation.ID,
		fileUploadPreparedRepairPoisoned,
	).Scan(&retryCount, &lastError); err != nil {
		t.Fatal(err)
	}
	if retryCount != 1 || lastError != "storage_authority_missing" {
		t.Fatalf("prepared poison = retry %d error %q", retryCount, lastError)
	}
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET StorageProvider = ?, StorageBucket = ?
WHERE Id = ?`,
		storage.ProviderFilesystem,
		storage.DefaultFilesystemBucket,
		operation.ID,
	); err != nil {
		t.Fatal(err)
	}
	authority, err := persistence.NewRuntimeWorkOwnershipRepository(db, ownership)
	if err != nil {
		t.Fatal(err)
	}
	repairStore, err := NewRuntimeUploadLifecycleRepairStore(db, authority)
	if err != nil {
		t.Fatal(err)
	}
	errorDigest := sha256.Sum256([]byte(lastError))
	if err := repairStore.Repair(t.Context(), persistence.RuntimePoisonRepair{
		RowID: operation.ID, ExpectedRetryCount: retryCount,
		ExpectedLastErrorSHA256: hex.EncodeToString(errorDigest[:]),
		RepairID:                "repair-prepared-authority", Reason: "restore recorded authority",
	}); err != nil {
		t.Fatal(err)
	}
	assertRuntimeUploadJournalState(
		t,
		db,
		operation.ID,
		persistence.FileUploadOperationFailed,
		fileUploadPreparedCleanupClaimed,
	)
	if processed, err := lifecycle.ProcessExpired(
		t.Context(),
		1,
		time.Minute,
	); err != nil || processed != 1 {
		t.Fatalf("repaired first pass = %d, %v", processed, err)
	}
	assertRuntimeUploadJournalState(
		t,
		db,
		operation.ID,
		persistence.FileUploadOperationFailed,
		fileUploadPreparedCleanupQuiet,
	)
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET ExpiresAt = DATE_SUB(UTC_TIMESTAMP(6), INTERVAL 1 SECOND)
WHERE Id = ?`, operation.ID); err != nil {
		t.Fatal(err)
	}
	if processed, err := lifecycle.ProcessExpired(
		t.Context(),
		1,
		time.Minute,
	); err != nil || processed != 1 {
		t.Fatalf("repaired final pass = %d, %v", processed, err)
	}
	assertRuntimeUploadJournalAbsent(t, db, operation.ID)
}

func TestRuntimeUploadLifecycleTiDBNullObjectKeyPoisonsWithoutStallingBatch(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	objectStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	ownership := activateRuntimeUploadLifecycleOwner(
		t,
		db,
		"runtime-files-null-object-key",
		1,
	)
	lifecycle, err := NewRuntimeUploadLifecycle(db, objectStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	malformed := seedRuntimeUploadPreparedOperation(
		t,
		db,
		"00-null-object-key",
		storage.DefaultFilesystemBucket,
	)
	valid := seedRuntimeUploadPreparedOperation(
		t,
		db,
		"01-valid-after-null",
		storage.DefaultFilesystemBucket,
	)
	putRuntimeUploadObject(t, objectStore, valid.ObjectKey, "valid")
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET ObjectKey = NULL, ExpiresAt = DATE_SUB(UTC_TIMESTAMP(6), INTERVAL 2 SECOND)
WHERE Id = ?`, malformed.ID); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET ExpiresAt = DATE_SUB(UTC_TIMESTAMP(6), INTERVAL 1 SECOND)
WHERE Id = ?`, valid.ID); err != nil {
		t.Fatal(err)
	}

	processed, err := lifecycle.ProcessExpired(t.Context(), 2, time.Minute)
	if err != nil || processed != 2 {
		t.Fatalf("process batch = %d, %v", processed, err)
	}
	assertRuntimeUploadJournalState(
		t,
		db,
		malformed.ID,
		persistence.FileUploadOperationFailed,
		fileUploadPreparedRepairPoisoned,
	)
	var poisonMessage string
	if err := db.QueryRow(`SELECT ErrorMessage
FROM _System_FileUploadOperation WHERE Id = ?`, malformed.ID).Scan(&poisonMessage); err != nil {
		t.Fatal(err)
	}
	if poisonMessage != "journal_identity_invalid" {
		t.Fatalf("nullable object-key poison = %q", poisonMessage)
	}
	assertRuntimeUploadJournalState(
		t,
		db,
		valid.ID,
		persistence.FileUploadOperationFailed,
		fileUploadPreparedCleanupQuiet,
	)
	if _, err := objectStore.HeadObject(t.Context(), valid.ObjectKey); !errors.Is(
		err,
		storage.ErrObjectNotFound,
	) {
		t.Fatalf("valid later claim was not processed: %v", err)
	}
}

func TestRuntimeUploadLifecycleTiDBNullChecksumPoisonsWithoutParkingClaim(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	objectStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	ownership := activateRuntimeUploadLifecycleOwner(t, db, "runtime-files-null-checksum", 1)
	lifecycle, err := NewRuntimeUploadLifecycle(db, objectStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	operation := seedRuntimeUploadLifecycleOperation(
		t, db, objectStore, "null-checksum", true, true,
	)
	if _, err := db.Exec(
		`UPDATE _System_File SET Checksum = NULL WHERE Id = ?`,
		operation.SystemFileID,
	); err != nil {
		t.Fatal(err)
	}

	processed, err := lifecycle.ProcessExpired(t.Context(), 1, time.Minute)
	if err != nil || processed != 1 {
		t.Fatalf("process = %d, %v", processed, err)
	}
	var state, errorCode, errorMessage string
	if err := db.QueryRow(`SELECT State, ErrorCode, ErrorMessage
FROM _System_FileUploadOperation WHERE Id = ?`, operation.ID).Scan(
		&state, &errorCode, &errorMessage,
	); err != nil {
		t.Fatal(err)
	}
	if state != persistence.FileUploadOperationFailed ||
		errorCode != fileUploadRepairPoisoned || errorMessage != "metadata_mismatch" {
		t.Fatalf("nullable checksum disposition = %q/%q/%q", state, errorCode, errorMessage)
	}
}

func TestRuntimeUploadLifecycleTiDBExactPoisonRepairRevalidatesSameObligation(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	objectStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	ownership := activateRuntimeUploadLifecycleOwner(t, db, "runtime-files-repair", 1)
	lifecycle, err := NewRuntimeUploadLifecycle(db, objectStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	operation := seedRuntimeUploadLifecycleOperation(
		t, db, objectStore, "repairable", true, false,
	)
	if processed, err := lifecycle.ProcessExpired(t.Context(), 1, time.Minute); err != nil || processed != 1 {
		t.Fatalf("poison pass = processed %d error %v", processed, err)
	}

	var retryCount int
	var lastError string
	if err := db.QueryRow(`SELECT RetryCount, ErrorMessage
FROM _System_FileUploadOperation WHERE Id = ?`, operation.ID).Scan(
		&retryCount, &lastError,
	); err != nil {
		t.Fatal(err)
	}
	if retryCount != 1 || lastError != "committed_blob_missing" {
		t.Fatalf("poison observation = retry %d error %q", retryCount, lastError)
	}
	authority, err := persistence.NewRuntimeWorkOwnershipRepository(db, ownership)
	if err != nil {
		t.Fatal(err)
	}
	repairStore, err := NewRuntimeUploadLifecycleRepairStore(db, authority)
	if err != nil {
		t.Fatal(err)
	}
	if err := repairStore.CheckReady(t.Context()); err != nil {
		t.Fatal(err)
	}
	wrongDigest := sha256.Sum256([]byte("different observation"))
	if err := repairStore.Repair(t.Context(), persistence.RuntimePoisonRepair{
		RowID: operation.ID, ExpectedRetryCount: retryCount,
		ExpectedLastErrorSHA256: hex.EncodeToString(wrongDigest[:]),
		RepairID:                "repair-files-wrong", Reason: "must not match",
	}); err == nil || !strings.Contains(err.Error(), "observation changed") {
		t.Fatalf("changed poison observation was accepted: %v", err)
	}
	errorDigest := sha256.Sum256([]byte(lastError))
	firstRepair := persistence.RuntimePoisonRepair{
		RowID: operation.ID, ExpectedRetryCount: retryCount,
		ExpectedLastErrorSHA256: hex.EncodeToString(errorDigest[:]),
		RepairID:                "repair-files-without-blob", Reason: "revalidate exact immutable blob",
	}
	if err := repairStore.Repair(t.Context(), firstRepair); err != nil {
		t.Fatal(err)
	}
	if err := repairStore.Repair(t.Context(), firstRepair); err == nil {
		t.Fatal("duplicate immediate repair was accepted")
	}
	if processed, err := lifecycle.ProcessExpired(t.Context(), 1, time.Minute); err != nil || processed != 1 {
		t.Fatalf("unrepaired blob revalidation = processed %d error %v", processed, err)
	}
	var state, repairID, repairReason string
	var repairCount int
	if err := db.QueryRow(`SELECT State, RetryCount, RepairId, RepairReason, RepairCount
FROM _System_FileUploadOperation WHERE Id = ?`, operation.ID).Scan(
		&state, &retryCount, &repairID, &repairReason, &repairCount,
	); err != nil {
		t.Fatal(err)
	}
	if state != persistence.FileUploadOperationFailed || retryCount != 2 ||
		repairID != "repair-files-without-blob" || repairReason != "revalidate exact immutable blob" ||
		repairCount != 1 {
		t.Fatalf(
			"re-poisoned row = state %q retry %d repair %q reason %q count %d",
			state, retryCount, repairID, repairReason, repairCount,
		)
	}
	if err := repairStore.Repair(t.Context(), firstRepair); err == nil ||
		!strings.Contains(err.Error(), "observation changed") {
		t.Fatalf("stale monotonic poison observation was accepted: %v", err)
	}

	body, _ := runtimeUploadFixture("repairable")
	size := int64(len(body))
	if _, err := objectStore.PutObject(
		t.Context(), operation.ObjectKey, strings.NewReader(body),
		storage.PutObjectOptions{ContentType: "text/plain", ContentLength: &size},
	); err != nil {
		t.Fatal(err)
	}
	if err := repairStore.Repair(t.Context(), persistence.RuntimePoisonRepair{
		RowID: operation.ID, ExpectedRetryCount: retryCount,
		ExpectedLastErrorSHA256: hex.EncodeToString(errorDigest[:]),
		RepairID:                "repair-files-restored", Reason: "restore exact immutable blob",
	}); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT State, RetryCount, RepairId, RepairReason, RepairCount
FROM _System_FileUploadOperation WHERE Id = ?`, operation.ID).Scan(
		&state, &retryCount, &repairID, &repairReason, &repairCount,
	); err != nil {
		t.Fatal(err)
	}
	if state != persistence.FileUploadOperationBlobStored || retryCount != 2 ||
		repairID != "repair-files-restored" || repairReason != "restore exact immutable blob" ||
		repairCount != 2 {
		t.Fatalf(
			"repaired row = state %q retry %d repair %q reason %q count %d",
			state, retryCount, repairID, repairReason, repairCount,
		)
	}
	if processed, err := lifecycle.ProcessExpired(t.Context(), 1, time.Minute); err != nil || processed != 1 {
		t.Fatalf("revalidation pass = processed %d error %v", processed, err)
	}
	if err := db.QueryRow(`SELECT State FROM _System_FileUploadOperation
WHERE Id = ?`, operation.ID).Scan(&state); err != nil {
		t.Fatal(err)
	}
	if state != persistence.FileUploadOperationCompleted {
		t.Fatalf("repaired obligation state = %q, want COMPLETED", state)
	}
}

func TestRuntimeUploadLifecycleTiDBRepairCancellationAndOwnershipTransferDoNotMutate(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	objectStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	oldOwnership := activateRuntimeUploadLifecycleOwner(t, db, "runtime-files-repair-old", 1)
	lifecycle, err := NewRuntimeUploadLifecycle(db, objectStore, oldOwnership)
	if err != nil {
		t.Fatal(err)
	}
	operation := seedRuntimeUploadLifecycleOperation(
		t, db, objectStore, "repair-fenced", true, false,
	)
	if processed, err := lifecycle.ProcessExpired(t.Context(), 1, time.Minute); err != nil || processed != 1 {
		t.Fatalf("poison pass = processed %d error %v", processed, err)
	}
	authority, err := persistence.NewRuntimeWorkOwnershipRepository(db, oldOwnership)
	if err != nil {
		t.Fatal(err)
	}
	repairStore, err := NewRuntimeUploadLifecycleRepairStore(db, authority)
	if err != nil {
		t.Fatal(err)
	}
	errorDigest := sha256.Sum256([]byte("committed_blob_missing"))
	repair := persistence.RuntimePoisonRepair{
		RowID: operation.ID, ExpectedRetryCount: 1,
		ExpectedLastErrorSHA256: hex.EncodeToString(errorDigest[:]),
		RepairID:                "repair-files-fenced", Reason: "must retain the old observation",
	}
	cancelled, cancel := context.WithCancel(context.Background())
	cancel()
	if err := repairStore.Repair(cancelled, repair); !errors.Is(err, context.Canceled) {
		t.Fatalf("cancelled repair error = %v, want context canceled", err)
	}
	assertRuntimeUploadRepairObservation(
		t, db, operation.ID, persistence.FileUploadOperationFailed, 1, "", 0,
	)

	newOwnership := persistence.RuntimeWorkOwnership{
		WorkFamily: oldOwnership.WorkFamily, DeploymentID: "runtime-files-repair-new",
		WorkerReleaseDigest: "sha256:" + strings.Repeat("b", 64),
		OwnershipEpoch:      2,
	}
	if _, err := db.Exec(`UPDATE _System_RuntimeWorkOwnership
SET DeploymentId = ?, WorkerReleaseDigest = ?, OwnershipEpoch = ?,
    OperationId = ?
WHERE WorkFamily = ? AND DeploymentId = ? AND OwnershipEpoch = 1`,
		newOwnership.DeploymentID, newOwnership.WorkerReleaseDigest,
		newOwnership.EpochDecimal(), "transfer-runtime-files-repair",
		newOwnership.WorkFamily, oldOwnership.DeploymentID,
	); err != nil {
		t.Fatal(err)
	}
	if err := repairStore.Repair(t.Context(), repair); !errors.Is(
		err, persistence.ErrRuntimeWorkOwnershipLost,
	) {
		t.Fatalf("stale-owner repair error = %v, want ownership lost", err)
	}
	assertRuntimeUploadRepairObservation(
		t, db, operation.ID, persistence.FileUploadOperationFailed, 1, "", 0,
	)
}

func assertRuntimeUploadRepairObservation(
	t *testing.T,
	db *sql.DB,
	id string,
	wantState string,
	wantRetry int,
	wantRepairID string,
	wantRepairCount int,
) {
	t.Helper()
	var state, repairID string
	var retryCount, repairCount int
	if err := db.QueryRow(`SELECT State, RetryCount, COALESCE(RepairId, ''), RepairCount
FROM _System_FileUploadOperation WHERE Id = ?`, id).Scan(
		&state, &retryCount, &repairID, &repairCount,
	); err != nil {
		t.Fatal(err)
	}
	if state != wantState || retryCount != wantRetry ||
		repairID != wantRepairID || repairCount != wantRepairCount {
		t.Fatalf(
			"repair observation = state %q retry %d repair %q count %d, want %q/%d/%q/%d",
			state, retryCount, repairID, repairCount,
			wantState, wantRetry, wantRepairID, wantRepairCount,
		)
	}
}

func TestRuntimeUploadLifecycleTiDBForegroundCommitWinsJournalFence(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	objectStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	ownership := activateRuntimeUploadLifecycleOwner(t, db, "runtime-files-foreground", 1)
	lifecycle, err := NewRuntimeUploadLifecycle(db, objectStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	operation := seedRuntimeUploadLifecycleOperation(
		t, db, objectStore, "foreground", false, true,
	)

	foregroundLocked := make(chan struct{})
	releaseForeground := make(chan struct{})
	foregroundResult := make(chan error, 1)
	go func() {
		foregroundResult <- completeRuntimeUploadForeground(
			context.Background(),
			db,
			operation,
			foregroundLocked,
			releaseForeground,
		)
	}()
	select {
	case <-foregroundLocked:
	case <-time.After(5 * time.Second):
		t.Fatal("foreground did not lock BLOB_STORED journal")
	}

	type processResult struct {
		processed int64
		err       error
	}
	workerResult := make(chan processResult, 1)
	go func() {
		processed, err := lifecycle.ProcessExpired(
			context.Background(), 1, time.Minute,
		)
		workerResult <- processResult{processed: processed, err: err}
	}()
	select {
	case result := <-workerResult:
		t.Fatalf(
			"worker crossed foreground journal fence: processed=%d err=%v",
			result.processed, result.err,
		)
	case <-time.After(150 * time.Millisecond):
	}
	close(releaseForeground)
	if err := <-foregroundResult; err != nil {
		t.Fatal(err)
	}
	result := <-workerResult
	if result.err != nil || result.processed != 0 {
		t.Fatalf(
			"worker after foreground commit = processed %d err %v, want no claim",
			result.processed, result.err,
		)
	}
	assertRuntimeUploadJournalState(
		t, db, operation.ID, persistence.FileUploadOperationCompleted, "",
	)
	var metadataCount int
	if err := db.QueryRow(
		`SELECT COUNT(*) FROM _System_File WHERE Id = ?`,
		operation.SystemFileID,
	).Scan(&metadataCount); err != nil {
		t.Fatal(err)
	}
	if metadataCount != 1 {
		t.Fatalf("foreground metadata count = %d, want 1", metadataCount)
	}
	if _, err := objectStore.HeadObject(
		t.Context(), operation.ObjectKey,
	); err != nil {
		t.Fatalf("foreground blob was removed: %v", err)
	}
}

func TestRuntimeUploadLifecycleTiDBWorkerClaimFencesForegroundCommit(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	baseStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	blockingStore := newBlockingDeleteObjectStore(baseStore)
	ownership := activateRuntimeUploadLifecycleOwner(t, db, "runtime-files-worker", 1)
	lifecycle, err := NewRuntimeUploadLifecycle(db, blockingStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	operation := seedRuntimeUploadLifecycleOperation(
		t, db, baseStore, "worker", false, true,
	)

	type processResult struct {
		processed int64
		err       error
	}
	workerResult := make(chan processResult, 1)
	go func() {
		processed, err := lifecycle.ProcessExpired(
			context.Background(), 1, time.Minute,
		)
		workerResult <- processResult{processed: processed, err: err}
	}()
	select {
	case <-blockingStore.started:
	case <-time.After(5 * time.Second):
		t.Fatal("worker did not claim journal before object cleanup")
	}

	var metadataCalled atomic.Bool
	err = persistence.NewFileUploadOperationRepository(db).FinalizeBlobStored(
		t.Context(),
		operation.ID,
		operation.OwnerToken,
		operation.RequestHash,
		operation.ObjectKey,
		time.Now().UTC().Add(time.Hour),
		func(context.Context) error {
			metadataCalled.Store(true)
			return nil
		},
	)
	if !errors.Is(err, persistence.ErrFileUploadOperationLostOwnership) {
		t.Fatalf("foreground after Worker claim = %v, want lost ownership", err)
	}
	if metadataCalled.Load() {
		t.Fatal("foreground metadata callback ran after Worker claimed journal")
	}
	close(blockingStore.release)
	result := <-workerResult
	if result.err != nil || result.processed != 1 {
		t.Fatalf(
			"worker cleanup = processed %d err %v, want one",
			result.processed, result.err,
		)
	}
	assertRuntimeUploadJournalAbsent(t, db, operation.ID)
	if _, err := baseStore.HeadObject(
		t.Context(), operation.ObjectKey,
	); !errors.Is(err, storage.ErrObjectNotFound) {
		t.Fatalf("Worker-owned orphan blob remains: %v", err)
	}
}

func TestRuntimeUploadLifecycleTiDBOwnershipTransferFencesOldDisposition(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	baseStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	blockingStore := newBlockingDeleteObjectStore(baseStore)
	oldOwnership := activateRuntimeUploadLifecycleOwner(t, db, "runtime-files-old", 1)
	oldLifecycle, err := NewRuntimeUploadLifecycle(db, blockingStore, oldOwnership)
	if err != nil {
		t.Fatal(err)
	}
	operation := seedRuntimeUploadLifecycleOperation(
		t, db, baseStore, "transfer", false, true,
	)

	type processResult struct {
		processed int64
		err       error
	}
	oldResult := make(chan processResult, 1)
	go func() {
		processed, err := oldLifecycle.ProcessExpired(
			context.Background(), 1, time.Minute,
		)
		oldResult <- processResult{processed: processed, err: err}
	}()
	select {
	case <-blockingStore.started:
	case <-time.After(5 * time.Second):
		t.Fatal("old Worker did not claim journal")
	}
	newOwnership := persistence.RuntimeWorkOwnership{
		WorkFamily:   RuntimeFileUploadLifecycleWorkFamily,
		DeploymentID: "runtime-files-new", WorkerReleaseDigest: "sha256:" + strings.Repeat("b", 64),
		OwnershipEpoch: 2,
	}
	if _, err := db.Exec(`UPDATE _System_RuntimeWorkOwnership
SET DeploymentId = ?, WorkerReleaseDigest = ?, OwnershipEpoch = ?,
    OperationId = ?
WHERE WorkFamily = ? AND DeploymentId = ? AND OwnershipEpoch = 1`,
		newOwnership.DeploymentID, newOwnership.WorkerReleaseDigest,
		newOwnership.EpochDecimal(), "transfer-runtime-files",
		newOwnership.WorkFamily, oldOwnership.DeploymentID,
	); err != nil {
		t.Fatal(err)
	}
	close(blockingStore.release)
	result := <-oldResult
	if !errors.Is(result.err, persistence.ErrRuntimeWorkOwnershipLost) ||
		result.processed != 0 {
		t.Fatalf(
			"old Worker disposition = processed %d err %v, want fenced",
			result.processed, result.err,
		)
	}
	assertRuntimeUploadJournalState(
		t, db, operation.ID, persistence.FileUploadOperationFailed,
		fileUploadCleanupClaimed,
	)

	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET ExpiresAt = DATE_SUB(UTC_TIMESTAMP(6), INTERVAL 1 SECOND)
WHERE Id = ?`, operation.ID); err != nil {
		t.Fatal(err)
	}
	newLifecycle, err := NewRuntimeUploadLifecycle(db, baseStore, newOwnership)
	if err != nil {
		t.Fatal(err)
	}
	processed, err := newLifecycle.ProcessExpired(
		t.Context(), 1, time.Minute,
	)
	if err != nil || processed != 1 {
		t.Fatalf(
			"new Worker restart cleanup = processed %d err %v",
			processed, err,
		)
	}
	assertRuntimeUploadJournalAbsent(t, db, operation.ID)
}

func TestRuntimeUploadLifecycleTiDBCancellationLeavesRestartableClaim(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	baseStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	blockingStore := newBlockingDeleteObjectStore(baseStore)
	ownership := activateRuntimeUploadLifecycleOwner(t, db, "runtime-files-cancel", 1)
	lifecycle, err := NewRuntimeUploadLifecycle(db, blockingStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	operation := seedRuntimeUploadLifecycleOperation(
		t, db, baseStore, "cancel", false, true,
	)

	ctx, cancel := context.WithCancel(context.Background())
	type processResult struct {
		processed int64
		err       error
	}
	resultChannel := make(chan processResult, 1)
	go func() {
		processed, err := lifecycle.ProcessExpired(ctx, 1, time.Minute)
		resultChannel <- processResult{processed: processed, err: err}
	}()
	select {
	case <-blockingStore.started:
	case <-time.After(5 * time.Second):
		t.Fatal("Worker did not reach cancellable object cleanup")
	}
	cancel()
	result := <-resultChannel
	if !errors.Is(result.err, context.Canceled) || result.processed != 0 {
		t.Fatalf(
			"cancelled Worker = processed %d err %v",
			result.processed, result.err,
		)
	}
	assertRuntimeUploadJournalState(
		t, db, operation.ID, persistence.FileUploadOperationFailed,
		fileUploadCleanupClaimed,
	)
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET ExpiresAt = DATE_SUB(UTC_TIMESTAMP(6), INTERVAL 1 SECOND)
WHERE Id = ?`, operation.ID); err != nil {
		t.Fatal(err)
	}
	restarted, err := NewRuntimeUploadLifecycle(db, baseStore, ownership)
	if err != nil {
		t.Fatal(err)
	}
	processed, err := restarted.ProcessExpired(
		t.Context(), 1, time.Minute,
	)
	if err != nil || processed != 1 {
		t.Fatalf(
			"restart after cancellation = processed %d err %v",
			processed, err,
		)
	}
	assertRuntimeUploadJournalAbsent(t, db, operation.ID)
}

func TestRuntimeUploadLifecycleTiDBTransientRowDoesNotStrandClaimedBatch(t *testing.T) {
	db, cleanup := openFileUploadTiDB(t)
	defer cleanup()
	createFileUploadOperationTable(t, db)
	createRuntimeUploadLifecycleTables(t, db)
	baseStore, err := storage.NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	transient := seedRuntimeUploadLifecycleOperation(
		t, db, baseStore, "a-transient", false, true,
	)
	orphan := seedRuntimeUploadLifecycleOperation(
		t, db, baseStore, "b-orphan", false, true,
	)
	completed := seedRuntimeUploadLifecycleOperation(
		t, db, baseStore, "c-complete", true, true,
	)
	sentinel := errors.New("transient object-store failure")
	store := &failDeleteForKeyObjectStore{
		ObjectStore: baseStore,
		key:         transient.ObjectKey,
		err:         sentinel,
	}
	ownership := activateRuntimeUploadLifecycleOwner(
		t, db, "runtime-files-batch", 1,
	)
	lifecycle, err := NewRuntimeUploadLifecycle(db, store, ownership)
	if err != nil {
		t.Fatal(err)
	}
	processed, err := lifecycle.ProcessExpired(
		t.Context(), 10, time.Minute,
	)
	if processed != 2 || !errors.Is(err, sentinel) {
		t.Fatalf(
			"batch with first transient = processed %d err %v, want 2/sentinel",
			processed, err,
		)
	}
	assertRuntimeUploadJournalState(
		t, db, transient.ID, persistence.FileUploadOperationFailed,
		fileUploadCleanupClaimed,
	)
	assertRuntimeUploadJournalAbsent(t, db, orphan.ID)
	assertRuntimeUploadJournalState(
		t, db, completed.ID, persistence.FileUploadOperationCompleted, "",
	)
}

type blockingDeleteObjectStore struct {
	storage.ObjectStore
	started chan struct{}
	release chan struct{}
	once    sync.Once
}

func newBlockingDeleteObjectStore(
	delegate storage.ObjectStore,
) *blockingDeleteObjectStore {
	return &blockingDeleteObjectStore{
		ObjectStore: delegate,
		started:     make(chan struct{}),
		release:     make(chan struct{}),
	}
}

func (s *blockingDeleteObjectStore) DeleteObject(
	ctx context.Context,
	key string,
) error {
	s.once.Do(func() { close(s.started) })
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-s.release:
		return s.ObjectStore.DeleteObject(ctx, key)
	}
}

type failDeleteForKeyObjectStore struct {
	storage.ObjectStore
	key string
	err error
}

func (s *failDeleteForKeyObjectStore) DeleteObject(
	ctx context.Context,
	key string,
) error {
	if key == s.key {
		return s.err
	}
	return s.ObjectStore.DeleteObject(ctx, key)
}

func completeRuntimeUploadForeground(
	ctx context.Context,
	db *sql.DB,
	operation persistence.FileUploadOperation,
	locked chan<- struct{},
	release <-chan struct{},
) error {
	tx, cleanup, err := persistence.BeginPessimisticTx(ctx, db)
	if err != nil {
		return err
	}
	defer func() { _ = cleanup() }()
	var state string
	if err := tx.QueryRowContext(ctx, `SELECT State
FROM _System_FileUploadOperation
WHERE Id = ? AND OwnerToken = ? AND RequestHash = ? AND ObjectKey = ?
FOR UPDATE`,
		operation.ID, operation.OwnerToken, operation.RequestHash,
		operation.ObjectKey,
	).Scan(&state); err != nil {
		return err
	}
	if state != persistence.FileUploadOperationBlobStored {
		return persistence.ErrFileUploadOperationLostOwnership
	}
	body, contentDigest := runtimeUploadFixture(
		strings.TrimPrefix(operation.ID, "operation-"),
	)
	fileName := strings.TrimPrefix(operation.ID, "operation-") + ".txt"
	if _, err := tx.ExecContext(ctx, `INSERT INTO _System_File
(Id, ParentId, ParentObjectApiName, Name, OriginalFileName, MimeType, SizeBytes,
 StoragePath, StorageKey, AttachmentKind, Checksum)
VALUES (?, ?, ?, ?, ?, 'text/plain', ?, ?, ?, ?, ?)`,
		operation.SystemFileID, operation.ParentID, operation.ParentObjectAPIName,
		fileName, fileName, len(body), operation.ObjectKey, operation.ObjectKey,
		defaultRecordAttachmentKind, contentDigest,
	); err != nil {
		return err
	}
	close(locked)
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-release:
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_FileUploadOperation
SET State = 'COMPLETED', ExpiresAt = DATE_ADD(UTC_TIMESTAMP(6), INTERVAL 24 HOUR)
WHERE Id = ? AND State = 'BLOB_STORED' AND OwnerToken = ?
  AND RequestHash = ? AND ObjectKey = ?`,
		operation.ID, operation.OwnerToken, operation.RequestHash,
		operation.ObjectKey,
	)
	if err != nil {
		return err
	}
	if affected, err := result.RowsAffected(); err != nil || affected != 1 {
		return persistence.ErrFileUploadOperationLostOwnership
	}
	return tx.Commit()
}

func assertRuntimeUploadJournalState(
	t *testing.T,
	db *sql.DB,
	operationID string,
	wantState string,
	wantErrorCode string,
) {
	t.Helper()
	var state, errorCode string
	if err := db.QueryRow(`SELECT State, COALESCE(ErrorCode, '')
FROM _System_FileUploadOperation WHERE Id = ?`, operationID).Scan(
		&state, &errorCode,
	); err != nil {
		t.Fatal(err)
	}
	if state != wantState || errorCode != wantErrorCode {
		t.Fatalf(
			"journal %s = state %q error %q, want %q/%q",
			operationID, state, errorCode, wantState, wantErrorCode,
		)
	}
}

func assertRuntimeUploadJournalAbsent(
	t *testing.T,
	db *sql.DB,
	operationID string,
) {
	t.Helper()
	var count int
	if err := db.QueryRow(
		`SELECT COUNT(*) FROM _System_FileUploadOperation WHERE Id = ?`,
		operationID,
	).Scan(&count); err != nil {
		t.Fatal(err)
	}
	if count != 0 {
		t.Fatalf("journal %s count = %d, want absent", operationID, count)
	}
}

func createRuntimeUploadLifecycleTables(t *testing.T, db *sql.DB) {
	t.Helper()
	if _, err := db.Exec(`CREATE TABLE _System_RuntimeWorkOwnership (
  WorkFamily VARCHAR(64) NOT NULL,
  DeploymentId VARCHAR(255) NOT NULL,
  WorkerReleaseDigest CHAR(71) NOT NULL,
  OwnershipEpoch DECIMAL(20,0) NOT NULL,
  IsActive TINYINT(1) NOT NULL,
  OperationId VARCHAR(255) NOT NULL,
  PRIMARY KEY (WorkFamily)
)`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`CREATE TABLE _System_File (
  Id VARCHAR(255) NOT NULL,
  ParentId VARCHAR(255) NOT NULL,
  ParentObjectApiName VARCHAR(255) NOT NULL,
  Name VARCHAR(255) NOT NULL,
  OriginalFileName VARCHAR(255) NOT NULL,
  MimeType VARCHAR(255) NOT NULL,
  SizeBytes BIGINT NOT NULL,
  StoragePath TEXT NOT NULL,
  StorageKey VARCHAR(1024) NOT NULL,
  StorageProvider VARCHAR(32) NULL,
  StorageBucket VARCHAR(255) NULL,
  Status VARCHAR(32) NOT NULL DEFAULT 'Active',
  SourceType VARCHAR(32) NOT NULL DEFAULT 'native',
  AttachmentKind VARCHAR(64) NOT NULL,
  Checksum VARCHAR(255) NULL,
  PRIMARY KEY (Id)
)`); err != nil {
		t.Fatal(err)
	}
}

func activateRuntimeUploadLifecycleOwner(
	t *testing.T,
	db *sql.DB,
	deploymentID string,
	epoch uint64,
) persistence.RuntimeWorkOwnership {
	t.Helper()
	ownership := persistence.RuntimeWorkOwnership{
		WorkFamily: RuntimeFileUploadLifecycleWorkFamily, DeploymentID: deploymentID,
		WorkerReleaseDigest: "sha256:" + strings.Repeat("a", 64), OwnershipEpoch: epoch,
	}
	if _, err := db.Exec(`INSERT INTO _System_RuntimeWorkOwnership
(WorkFamily, DeploymentId, WorkerReleaseDigest, OwnershipEpoch, IsActive, OperationId)
VALUES (?, ?, ?, ?, 1, ?)`,
		ownership.WorkFamily, ownership.DeploymentID, ownership.WorkerReleaseDigest,
		ownership.EpochDecimal(), "activate-"+deploymentID,
	); err != nil {
		t.Fatal(err)
	}
	return ownership
}

func seedRuntimeUploadLifecycleOperation(
	t *testing.T,
	db *sql.DB,
	store storage.ObjectStore,
	id string,
	withMetadata bool,
	withBlob bool,
) persistence.FileUploadOperation {
	t.Helper()
	body, contentDigest := runtimeUploadFixture(id)
	operation := persistence.FileUploadOperation{
		ID: "operation-" + id, IdempotencyKey: "request-" + id,
		OwnerToken:      "owner-" + id,
		StorageProvider: storage.ProviderFilesystem,
		StorageBucket:   storage.DefaultFilesystemBucket,
		SystemFileID:    "file-" + id, ParentID: "record-" + id, ParentObjectAPIName: "Widget",
	}
	config := storage.LoadConfigFromEnv()
	config.Provider = operation.StorageProvider
	config.Bucket = operation.StorageBucket
	storageAuthorityDigest, err := storage.ConfigAuthorityDigest(config)
	if err != nil {
		t.Fatal(err)
	}
	operation.StorageAuthorityDigest = storageAuthorityDigest
	logicalObjectKey := systemFileObjectKey(
		operation.ParentObjectAPIName,
		operation.ParentID,
		operation.SystemFileID,
		id+".txt",
	)
	operation.ObjectKey = durableUploadAttemptObjectKey(
		logicalObjectKey, operation.OwnerToken,
	)
	header, err := durableUploadHeaderHash(contract.FileUploadRequest{
		ID: operation.SystemFileID, FileName: id + ".txt", ContentType: "text/plain",
		ContentLength: int64(len(body)), ParentID: operation.ParentID,
		AttachmentKind: defaultRecordAttachmentKind,
	}, operation.ParentObjectAPIName, logicalObjectKey)
	if err != nil {
		t.Fatal(err)
	}
	operation.RequestHash = header + "|" + contentDigest
	if _, err := db.Exec(`INSERT INTO _System_FileUploadOperation
(Id, IdempotencyKey, RequestHash, State, OwnerToken, ObjectKey,
 StorageProvider, StorageBucket, StorageAuthorityDigest,
 SystemFileId, ParentId, ParentObjectApiName, ExpiresAt)
VALUES (?, ?, ?, 'BLOB_STORED', ?, ?, ?, ?, ?, ?, ?, ?, DATE_SUB(UTC_TIMESTAMP(6), INTERVAL 1 SECOND))`,
		operation.ID, operation.IdempotencyKey, operation.RequestHash, operation.OwnerToken,
		operation.ObjectKey, operation.StorageProvider, operation.StorageBucket,
		operation.StorageAuthorityDigest,
		operation.SystemFileID, operation.ParentID, operation.ParentObjectAPIName,
	); err != nil {
		t.Fatal(err)
	}
	if withMetadata {
		if _, err := db.Exec(`INSERT INTO _System_File
(Id, ParentId, ParentObjectApiName, Name, OriginalFileName, MimeType, SizeBytes,
 StoragePath, StorageKey, StorageProvider, StorageBucket, AttachmentKind, Checksum)
VALUES (?, ?, ?, ?, ?, 'text/plain', ?, ?, ?, ?, ?, ?, ?)`,
			operation.SystemFileID, operation.ParentID, operation.ParentObjectAPIName,
			id+".txt", id+".txt", len(body), operation.ObjectKey, operation.ObjectKey,
			operation.StorageProvider, operation.StorageBucket,
			defaultRecordAttachmentKind, contentDigest,
		); err != nil {
			t.Fatal(err)
		}
	}
	if withBlob {
		size := int64(len(body))
		if _, err := store.PutObject(
			t.Context(), operation.ObjectKey, strings.NewReader(body),
			storage.PutObjectOptions{ContentType: "text/plain", ContentLength: &size},
		); err != nil {
			t.Fatal(err)
		}
	}
	return operation
}

func seedRuntimeUploadPreparedOperation(
	t *testing.T,
	db *sql.DB,
	id string,
	bucket string,
) persistence.FileUploadOperation {
	t.Helper()
	now := time.Now().UTC().Truncate(time.Microsecond)
	operation := persistence.FileUploadOperation{
		ID: "operation-" + id, IdempotencyKey: "request-" + id,
		RequestHash: "sha256:" + strings.Repeat("a", 64),
		OwnerToken:  "owner-" + id,
		ObjectKey: path.Join(
			"system-file",
			"Widget",
			"record-"+id,
			"file-"+id,
			".attempts",
			"owner-"+id,
			id+".txt",
		),
		StorageProvider: storage.ProviderFilesystem,
		StorageBucket:   bucket,
		SystemFileID:    "file-" + id, ParentID: "record-" + id,
		ParentObjectAPIName: "Widget",
		ExpiresAt:           now.Add(-time.Minute),
	}
	config := storage.LoadConfigFromEnv()
	config.Provider = operation.StorageProvider
	config.Bucket = operation.StorageBucket
	var err error
	operation.StorageAuthorityDigest, err = storage.ConfigAuthorityDigest(config)
	if err != nil {
		t.Fatal(err)
	}
	acceptance := persistence.FileUploadOperationAcceptance{
		ID: operation.ID, IdempotencyKey: operation.IdempotencyKey,
		RequestHash: operation.RequestHash, OwnerToken: operation.OwnerToken,
		ObjectKey: operation.ObjectKey, StorageProvider: operation.StorageProvider,
		StorageBucket: operation.StorageBucket, SystemFileID: operation.SystemFileID,
		StorageAuthorityDigest: operation.StorageAuthorityDigest,
		ParentID:               operation.ParentID, ParentObjectAPIName: operation.ParentObjectAPIName,
		Now: now.Add(-2 * time.Minute), LeaseUntil: operation.ExpiresAt,
	}
	row, owner, err := persistence.NewFileUploadOperationRepository(db).Accept(
		t.Context(),
		acceptance,
	)
	if err != nil || !owner {
		t.Fatalf("seed prepared operation = owner %t error %v", owner, err)
	}
	return row
}

func putRuntimeUploadObject(
	t *testing.T,
	store storage.ObjectStore,
	key string,
	body string,
) {
	t.Helper()
	size := int64(len(body))
	if _, err := store.PutObject(
		t.Context(),
		key,
		strings.NewReader(body),
		storage.PutObjectOptions{ContentType: "text/plain", ContentLength: &size},
	); err != nil {
		t.Fatal(err)
	}
}

func runtimeUploadStorageAuthorityDigest(
	t *testing.T,
	base storage.Config,
	provider string,
	bucket string,
) string {
	t.Helper()
	base.Provider = provider
	base.Bucket = bucket
	digest, err := storage.ConfigAuthorityDigest(base)
	if err != nil {
		t.Fatal(err)
	}
	return digest
}

func assertRuntimeUploadObjectBody(
	t *testing.T,
	store storage.ObjectStore,
	key string,
	want string,
) {
	t.Helper()
	reader, _, err := store.GetObject(t.Context(), key)
	if err != nil {
		t.Fatal(err)
	}
	body, readErr := io.ReadAll(reader)
	closeErr := reader.Close()
	if readErr != nil || closeErr != nil || string(body) != want {
		t.Fatalf(
			"object %q = %q read=%v close=%v, want %q",
			key,
			body,
			readErr,
			closeErr,
			want,
		)
	}
}

func runtimeUploadFixture(id string) (string, string) {
	body := "body-" + id
	digestBytes := sha256.Sum256([]byte(body))
	return body, "sha256:" + hex.EncodeToString(digestBytes[:])
}