package files

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"path"
	"strings"
	"time"

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

const (
	RuntimeFileUploadLifecycleWorkFamily = "runtime.file-upload-lifecycle.v1"

	fileUploadCleanupClaimed              = "BLOB_CLEANUP_CLAIMED"
	fileUploadPreparedCleanupClaimed      = "PREPARED_CLEANUP_CLAIMED"
	fileUploadPreparedCleanupQuiet        = "PREPARED_CLEANUP_QUIET"
	fileUploadPreparedCleanupFinalClaimed = "PREPARED_CLEANUP_FINAL_CLAIMED"
	fileUploadRepairPoisoned              = "BLOB_REPAIR_POISONED"
	fileUploadPreparedRepairPoisoned      = "PREPARED_REPAIR_POISONED"
)

var errDurableUploadStorageAuthorityInvalid = errors.New(
	"durable file upload storage authority is invalid",
)
var errDurableUploadStorageAuthorityMismatch = errors.New(
	"durable file upload storage authority does not match the current deployment mapping",
)

type RuntimeUploadLifecycleState struct {
	QueueDepth  int64
	PoisonDepth int64
	OldestAge   time.Duration
}

type RuntimeUploadLifecycleObserver struct {
	db *sql.DB
}

type runtimeUploadClaimMode uint8

const (
	runtimeUploadClaimBlobStored runtimeUploadClaimMode = iota + 1
	runtimeUploadClaimPreparedFirstPass
	runtimeUploadClaimPreparedFinalPass
)

type RuntimeUploadLifecycle struct {
	db                   *sql.DB
	store                storage.ObjectStore
	defaultStorageConfig storage.Config
	resolveStore         func(storage.Config) (storage.ObjectStore, error)
	ownership            *persistence.RuntimeWorkOwnershipRepository
}

func NewRuntimeUploadLifecycle(
	db *sql.DB,
	store storage.ObjectStore,
	expected persistence.RuntimeWorkOwnership,
) (*RuntimeUploadLifecycle, error) {
	if expected.WorkFamily != RuntimeFileUploadLifecycleWorkFamily {
		return nil, fmt.Errorf("runtime file upload lifecycle: wrong work family %q", expected.WorkFamily)
	}
	if db == nil || store == nil {
		return nil, errors.New("runtime file upload lifecycle: database and object store are required")
	}
	ownership, err := persistence.NewRuntimeWorkOwnershipRepository(db, expected)
	if err != nil {
		return nil, err
	}
	return &RuntimeUploadLifecycle{
		db:                   db,
		store:                store,
		defaultStorageConfig: storage.LoadConfigFromEnv(),
		resolveStore:         storage.NewStore,
		ownership:            ownership,
	}, nil
}

func (r *RuntimeUploadLifecycle) CheckReady(ctx context.Context) error {
	if r == nil || r.db == nil || r.store == nil || r.ownership == nil {
		return errors.New("runtime file upload lifecycle: runtime is unavailable")
	}
	if err := r.ownership.CheckReady(ctx); err != nil {
		return err
	}
	if err := persistence.VerifyFileUploadOperationSchema(ctx, r.db); err != nil {
		return err
	}
	return persistence.VerifyFileUploadLifecycleIndex(ctx, r.db)
}

func NewRuntimeUploadLifecycleObserver(db *sql.DB) (*RuntimeUploadLifecycleObserver, error) {
	if db == nil {
		return nil, errors.New("runtime file upload lifecycle observer: database is required")
	}
	return &RuntimeUploadLifecycleObserver{db: db}, nil
}

func (o *RuntimeUploadLifecycleObserver) Observe(ctx context.Context) (RuntimeUploadLifecycleState, error) {
	if o == nil || o.db == nil {
		return RuntimeUploadLifecycleState{}, errors.New("runtime file upload lifecycle: runtime is unavailable")
	}
	var state RuntimeUploadLifecycleState
	var oldestMicros int64
	err := o.db.QueryRowContext(ctx, `SELECT
  COALESCE(SUM(CASE WHEN ExpiresAt <= UTC_TIMESTAMP(6) AND
      (State IN ('PREPARED', 'BLOB_STORED') OR
       (State = 'FAILED' AND ErrorCode IN (?, ?, ?, ?)))
      THEN 1 ELSE 0 END), 0),
  COALESCE(SUM(CASE WHEN State = 'FAILED' AND ErrorCode IN (?, ?)
      THEN 1 ELSE 0 END), 0),
  COALESCE(TIMESTAMPDIFF(MICROSECOND,
    MIN(CASE WHEN ExpiresAt <= UTC_TIMESTAMP(6) AND
      (State IN ('PREPARED', 'BLOB_STORED') OR
       (State = 'FAILED' AND ErrorCode IN (?, ?, ?, ?)))
      THEN ExpiresAt END), UTC_TIMESTAMP(6)), 0)
FROM _System_FileUploadOperation`,
		fileUploadCleanupClaimed,
		fileUploadPreparedCleanupClaimed,
		fileUploadPreparedCleanupQuiet,
		fileUploadPreparedCleanupFinalClaimed,
		fileUploadRepairPoisoned,
		fileUploadPreparedRepairPoisoned,
		fileUploadCleanupClaimed,
		fileUploadPreparedCleanupClaimed,
		fileUploadPreparedCleanupQuiet,
		fileUploadPreparedCleanupFinalClaimed,
	).Scan(&state.QueueDepth, &state.PoisonDepth, &oldestMicros)
	if err != nil {
		return RuntimeUploadLifecycleState{}, fmt.Errorf("runtime file upload lifecycle: observe: %w", err)
	}
	if state.QueueDepth < 0 || state.PoisonDepth < 0 || oldestMicros < 0 {
		return RuntimeUploadLifecycleState{}, errors.New("runtime file upload lifecycle: invalid aggregate")
	}
	state.OldestAge = time.Duration(oldestMicros) * time.Microsecond
	return state, nil
}

// HasExpired is a bounded admission hint. It cannot authorize or suppress a
// durable transition: a positive result is rechecked by claim, while a read
// error must fall through to ProcessExpired in the Runtime Worker.
func (r *RuntimeUploadLifecycle) HasExpired(ctx context.Context) (bool, error) {
	if r == nil || r.db == nil {
		return false, errors.New("runtime file upload lifecycle: runtime is unavailable")
	}
	var exists bool
	err := r.db.QueryRowContext(ctx, `SELECT EXISTS(
  SELECT 1
  FROM _System_FileUploadOperation USE INDEX (idx__System_FileUploadOperation_lifecycle)
  WHERE ExpiresAt <= UTC_TIMESTAMP(6)
    AND (State IN ('PREPARED', 'BLOB_STORED') OR
         (State = 'FAILED' AND ErrorCode IN (?, ?, ?, ?)))
  LIMIT 1
)`,
		fileUploadCleanupClaimed,
		fileUploadPreparedCleanupClaimed,
		fileUploadPreparedCleanupQuiet,
		fileUploadPreparedCleanupFinalClaimed,
	).Scan(&exists)
	if err != nil {
		return false, fmt.Errorf("runtime file upload lifecycle: check expired: %w", err)
	}
	return exists, nil
}

type runtimeUploadClaim struct {
	operation  persistence.FileUploadOperation
	claimToken string
	claimCode  string
	mode       runtimeUploadClaimMode
}

// ProcessExpired claims a bounded set under the work-family fence, performs
// object-store I/O outside the database transaction, and then applies one
// separately fenced disposition per row. Object deletion is idempotent, so a
// lost acknowledgement or ownership transfer can safely repeat it.
func (r *RuntimeUploadLifecycle) ProcessExpired(
	ctx context.Context,
	batchSize int,
	leaseDuration time.Duration,
) (int64, error) {
	if r == nil || r.db == nil || r.store == nil || r.ownership == nil {
		return 0, errors.New("runtime file upload lifecycle: runtime is unavailable")
	}
	if batchSize <= 0 || batchSize > 10_000 || leaseDuration <= 0 {
		return 0, errors.New("runtime file upload lifecycle: invalid processing bounds")
	}
	claims, err := r.claim(ctx, batchSize, leaseDuration)
	if err != nil {
		return 0, err
	}
	var processed int64
	var failures error
	for _, claim := range claims {
		if err := ctx.Err(); err != nil {
			return processed, err
		}
		if err := r.processClaim(ctx, claim); err != nil {
			if errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) ||
				errors.Is(err, context.Canceled) ||
				errors.Is(err, context.DeadlineExceeded) {
				return processed, err
			}
			failures = errors.Join(failures, err)
			continue
		}
		processed++
	}
	return processed, failures
}

func (r *RuntimeUploadLifecycle) claim(
	ctx context.Context,
	batchSize int,
	leaseDuration time.Duration,
) ([]runtimeUploadClaim, error) {
	tx, cleanup, err := persistence.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return nil, fmt.Errorf("runtime file upload lifecycle: begin claim: %w", err)
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) ([]runtimeUploadClaim, error) {
		if rollbackErr := tx.Rollback(); rollbackErr != nil {
			return nil, errors.Join(cause, fmt.Errorf("runtime file upload lifecycle: rollback claim: %w", rollbackErr))
		}
		return nil, cause
	}
	if err := r.ownership.LockActive(ctx, tx); err != nil {
		return rollback(err)
	}
	var now time.Time
	if err := tx.QueryRowContext(ctx, `SELECT UTC_TIMESTAMP(6)`).Scan(&now); err != nil {
		return rollback(fmt.Errorf("runtime file upload lifecycle: read claim time: %w", err))
	}
	rows, err := tx.QueryContext(ctx, `SELECT Id, IdempotencyKey, RequestHash, State,
       OwnerToken, COALESCE(ObjectKey, ''), COALESCE(StorageProvider, ''),
       COALESCE(StorageBucket, ''), COALESCE(StorageAuthorityDigest, ''),
       COALESCE(SystemFileId, ''),
       ParentId, ParentObjectApiName, ExpiresAt, COALESCE(ErrorCode, '')
FROM _System_FileUploadOperation
WHERE ExpiresAt <= ?
  AND (State IN ('PREPARED', 'BLOB_STORED') OR
       (State = 'FAILED' AND ErrorCode IN (?, ?, ?, ?)))
ORDER BY ExpiresAt, Id
LIMIT ?
FOR UPDATE`,
		now,
		fileUploadCleanupClaimed,
		fileUploadPreparedCleanupClaimed,
		fileUploadPreparedCleanupQuiet,
		fileUploadPreparedCleanupFinalClaimed,
		batchSize,
	)
	if err != nil {
		return rollback(fmt.Errorf("runtime file upload lifecycle: select claims: %w", err))
	}
	type candidate struct {
		operation persistence.FileUploadOperation
		errorCode string
	}
	candidates := make([]candidate, 0, batchSize)
	for rows.Next() {
		var row persistence.FileUploadOperation
		var errorCode string
		if err := rows.Scan(
			&row.ID, &row.IdempotencyKey, &row.RequestHash, &row.State,
			&row.OwnerToken, &row.ObjectKey, &row.StorageProvider, &row.StorageBucket,
			&row.StorageAuthorityDigest,
			&row.SystemFileID,
			&row.ParentID, &row.ParentObjectAPIName, &row.ExpiresAt, &errorCode,
		); err != nil {
			_ = rows.Close()
			return rollback(fmt.Errorf("runtime file upload lifecycle: scan claim: %w", err))
		}
		candidates = append(candidates, candidate{operation: row, errorCode: errorCode})
	}
	if err := rows.Close(); err != nil {
		return rollback(fmt.Errorf("runtime file upload lifecycle: close claims: %w", err))
	}
	if err := rows.Err(); err != nil {
		return rollback(fmt.Errorf("runtime file upload lifecycle: iterate claims: %w", err))
	}
	claimUntil := now.Add(leaseDuration)
	claims := make([]runtimeUploadClaim, 0, len(candidates))
	for _, candidate := range candidates {
		claimCode, mode, err := classifyRuntimeUploadClaim(
			candidate.operation.State,
			candidate.errorCode,
		)
		if err != nil {
			return rollback(err)
		}
		token, err := newRuntimeUploadClaimToken()
		if err != nil {
			return rollback(err)
		}
		result, err := tx.ExecContext(ctx, `UPDATE _System_FileUploadOperation
SET State = 'FAILED', ErrorCode = ?, ErrorMessage = ?, ExpiresAt = ?,
    LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND OwnerToken = ? AND RequestHash = ?
  AND COALESCE(ObjectKey, '') = ?
  AND COALESCE(StorageProvider, '') = ?
  AND COALESCE(StorageBucket, '') = ?
  AND COALESCE(StorageAuthorityDigest, '') = ?
  AND ExpiresAt <= ?
  AND State = ? AND COALESCE(ErrorCode, '') = ?`,
			claimCode, token, claimUntil,
			candidate.operation.ID,
			candidate.operation.OwnerToken,
			candidate.operation.RequestHash,
			candidate.operation.ObjectKey,
			candidate.operation.StorageProvider,
			candidate.operation.StorageBucket,
			candidate.operation.StorageAuthorityDigest,
			now,
			candidate.operation.State,
			candidate.errorCode,
		)
		if err != nil {
			return rollback(fmt.Errorf(
				"runtime file upload lifecycle: claim %s: %w",
				candidate.operation.ID,
				err,
			))
		}
		affected, err := result.RowsAffected()
		if err != nil {
			return rollback(fmt.Errorf("runtime file upload lifecycle: claim rows affected: %w", err))
		}
		if affected != 1 {
			return rollback(errors.New("runtime file upload lifecycle: locked claim lost ownership"))
		}
		claims = append(claims, runtimeUploadClaim{
			operation:  candidate.operation,
			claimToken: token,
			claimCode:  claimCode,
			mode:       mode,
		})
	}
	if err := tx.Commit(); err != nil {
		return nil, fmt.Errorf("runtime file upload lifecycle: claim commit outcome unknown: %w", err)
	}
	return claims, nil
}

func classifyRuntimeUploadClaim(
	state string,
	errorCode string,
) (string, runtimeUploadClaimMode, error) {
	switch {
	case state == persistence.FileUploadOperationPrepared && errorCode == "":
		return fileUploadPreparedCleanupClaimed, runtimeUploadClaimPreparedFirstPass, nil
	case state == persistence.FileUploadOperationFailed &&
		errorCode == fileUploadPreparedCleanupClaimed:
		return fileUploadPreparedCleanupClaimed, runtimeUploadClaimPreparedFirstPass, nil
	case state == persistence.FileUploadOperationFailed &&
		(errorCode == fileUploadPreparedCleanupQuiet ||
			errorCode == fileUploadPreparedCleanupFinalClaimed):
		return fileUploadPreparedCleanupFinalClaimed, runtimeUploadClaimPreparedFinalPass, nil
	case state == persistence.FileUploadOperationBlobStored && errorCode == "":
		return fileUploadCleanupClaimed, runtimeUploadClaimBlobStored, nil
	case state == persistence.FileUploadOperationFailed && errorCode == fileUploadCleanupClaimed:
		return fileUploadCleanupClaimed, runtimeUploadClaimBlobStored, nil
	default:
		return "", 0, fmt.Errorf(
			"runtime file upload lifecycle: unsupported claim state %q/%q",
			state,
			errorCode,
		)
	}
}

type runtimeFileMetadata struct {
	id, parentID, parentObject, originalName, mimeType string
	storagePath, storageKey, storageProvider           string
	storageBucket, attachmentKind, checksum            string
	size                                               int64
}

func (r *RuntimeUploadLifecycle) processClaim(ctx context.Context, claim runtimeUploadClaim) error {
	if err := validateRuntimeUploadOperation(claim.operation, claim.mode); err != nil {
		if errors.Is(err, errDurableUploadStorageAuthorityInvalid) {
			return r.poisonClaim(ctx, claim, "storage_authority_missing")
		}
		return r.poisonClaim(ctx, claim, "journal_identity_invalid")
	}
	if claim.mode == runtimeUploadClaimPreparedFirstPass ||
		claim.mode == runtimeUploadClaimPreparedFinalPass {
		return r.processPreparedClaim(ctx, claim)
	}
	metadata, found, err := r.loadMetadata(ctx, claim.operation.SystemFileID)
	if err != nil {
		return err
	}
	if !found {
		if claim.operation.ObjectKey != "" {
			store, storeErr := r.storeForAuthority(
				claim.operation.StorageProvider,
				claim.operation.StorageBucket,
				claim.operation.StorageAuthorityDigest,
			)
			if storeErr != nil {
				if errors.Is(storeErr, errDurableUploadStorageAuthorityMismatch) {
					return r.poisonClaim(ctx, claim, "storage_authority_mismatch")
				}
				if errors.Is(storeErr, errDurableUploadStorageAuthorityInvalid) {
					return r.poisonClaim(ctx, claim, "storage_authority_missing")
				}
				return fmt.Errorf(
					"runtime file upload lifecycle: resolve orphan blob authority: %w",
					storeErr,
				)
			}
			if err := store.DeleteObject(ctx, claim.operation.ObjectKey); err != nil &&
				!errors.Is(err, storage.ErrObjectNotFound) {
				return fmt.Errorf("runtime file upload lifecycle: delete orphan blob: %w", err)
			}
		}
		return r.deleteClaim(ctx, claim)
	}
	contentDigest, err := validateRuntimeUploadMetadata(claim.operation, metadata)
	if err != nil {
		return r.poisonClaim(ctx, claim, "metadata_mismatch")
	}
	store, err := r.storeForAuthority(
		metadata.storageProvider,
		metadata.storageBucket,
		claim.operation.StorageAuthorityDigest,
	)
	if err != nil {
		if errors.Is(err, errDurableUploadStorageAuthorityMismatch) {
			return r.poisonClaim(ctx, claim, "storage_authority_mismatch")
		}
		if errors.Is(err, errDurableUploadStorageAuthorityInvalid) {
			return r.poisonClaim(ctx, claim, "storage_authority_missing")
		}
		return fmt.Errorf("runtime file upload lifecycle: resolve committed blob authority: %w", err)
	}
	reader, objectMetadata, err := store.GetObject(ctx, claim.operation.ObjectKey)
	if errors.Is(err, storage.ErrObjectNotFound) {
		return r.poisonClaim(ctx, claim, "committed_blob_missing")
	}
	if err != nil {
		return fmt.Errorf("runtime file upload lifecycle: open committed blob: %w", err)
	}
	hasher := sha256.New()
	count, copyErr := io.Copy(hasher, reader)
	closeErr := reader.Close()
	if copyErr != nil {
		return fmt.Errorf("runtime file upload lifecycle: read committed blob: %w", copyErr)
	}
	if closeErr != nil {
		return fmt.Errorf("runtime file upload lifecycle: close committed blob: %w", closeErr)
	}
	actualDigest := "sha256:" + hex.EncodeToString(hasher.Sum(nil))
	if count != metadata.size || objectMetadata.Key != claim.operation.ObjectKey ||
		objectMetadata.Size != metadata.size || actualDigest != contentDigest {
		return r.poisonClaim(ctx, claim, "committed_blob_mismatch")
	}
	return r.completeClaim(ctx, claim)
}

func validateRuntimeUploadOperation(
	operation persistence.FileUploadOperation,
	mode runtimeUploadClaimMode,
) error {
	if err := validateDurableUploadStorageAuthority(
		operation.StorageProvider,
		operation.StorageBucket,
	); err != nil {
		return err
	}
	if strings.TrimSpace(operation.ID) == "" ||
		strings.TrimSpace(operation.IdempotencyKey) == "" ||
		strings.TrimSpace(operation.OwnerToken) == "" ||
		strings.TrimSpace(operation.ObjectKey) == "" ||
		!validSHA256Digest(operation.StorageAuthorityDigest) ||
		strings.TrimSpace(operation.SystemFileID) == "" ||
		strings.TrimSpace(operation.ParentID) == "" ||
		strings.TrimSpace(operation.ParentObjectAPIName) == "" {
		return errors.New("runtime file upload lifecycle: journal identity is incomplete")
	}
	logicalObjectKey := path.Join(
		"system-file",
		operation.ParentObjectAPIName,
		operation.ParentID,
		operation.SystemFileID,
		path.Base(operation.ObjectKey),
	)
	if err := validateDurableUploadAttemptIdentity(
		operation,
		logicalObjectKey,
	); err != nil {
		return errors.New("runtime file upload lifecycle: journal object authority is invalid")
	}
	switch mode {
	case runtimeUploadClaimPreparedFirstPass, runtimeUploadClaimPreparedFinalPass:
		if !validSHA256Digest(operation.RequestHash) {
			return errors.New("runtime file upload lifecycle: prepared request identity is invalid")
		}
	case runtimeUploadClaimBlobStored:
		separator := strings.LastIndex(operation.RequestHash, "|")
		if separator <= 0 ||
			!validSHA256Digest(operation.RequestHash[:separator]) ||
			!validSHA256Digest(operation.RequestHash[separator+1:]) {
			return errors.New("runtime file upload lifecycle: stored request identity is invalid")
		}
	default:
		return errors.New("runtime file upload lifecycle: claim mode is invalid")
	}
	return nil
}

func (r *RuntimeUploadLifecycle) processPreparedClaim(
	ctx context.Context,
	claim runtimeUploadClaim,
) error {
	if strings.TrimSpace(claim.operation.ObjectKey) == "" {
		return r.poisonClaim(ctx, claim, "prepared_object_key_missing")
	}
	if _, found, err := r.loadMetadata(ctx, claim.operation.SystemFileID); err != nil {
		return err
	} else if found {
		return r.poisonClaim(ctx, claim, "prepared_metadata_conflict")
	}
	store, err := r.storeForAuthority(
		claim.operation.StorageProvider,
		claim.operation.StorageBucket,
		claim.operation.StorageAuthorityDigest,
	)
	if err != nil {
		if errors.Is(err, errDurableUploadStorageAuthorityMismatch) {
			return r.poisonClaim(ctx, claim, "storage_authority_mismatch")
		}
		if errors.Is(err, errDurableUploadStorageAuthorityInvalid) {
			return r.poisonClaim(ctx, claim, "storage_authority_missing")
		}
		return fmt.Errorf(
			"runtime file upload lifecycle: resolve prepared blob authority: %w",
			err,
		)
	}
	if err := store.DeleteObject(ctx, claim.operation.ObjectKey); err != nil &&
		!errors.Is(err, storage.ErrObjectNotFound) {
		return fmt.Errorf("runtime file upload lifecycle: delete prepared blob: %w", err)
	}
	if claim.mode == runtimeUploadClaimPreparedFirstPass {
		return r.startPreparedQuietPeriod(ctx, claim)
	}
	return r.deleteClaim(ctx, claim)
}

func (r *RuntimeUploadLifecycle) loadMetadata(
	ctx context.Context,
	fileID string,
) (runtimeFileMetadata, bool, error) {
	if strings.TrimSpace(fileID) == "" {
		return runtimeFileMetadata{}, false, nil
	}
	var row runtimeFileMetadata
	err := r.db.QueryRowContext(ctx, `SELECT Id, ParentId, ParentObjectApiName,
       OriginalFileName, MimeType, SizeBytes, StoragePath, StorageKey,
       COALESCE(StorageProvider, ''), COALESCE(StorageBucket, ''),
       AttachmentKind, COALESCE(Checksum, '')
FROM _System_File WHERE Id = ?
LIMIT 1`, fileID).Scan(
		&row.id, &row.parentID, &row.parentObject, &row.originalName,
		&row.mimeType, &row.size, &row.storagePath, &row.storageKey,
		&row.storageProvider, &row.storageBucket, &row.attachmentKind, &row.checksum,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return runtimeFileMetadata{}, false, nil
	}
	if err != nil {
		return runtimeFileMetadata{}, false, fmt.Errorf("runtime file upload lifecycle: load file metadata: %w", err)
	}
	return row, true, nil
}

func (r *RuntimeUploadLifecycle) storeForAuthority(
	storageProvider string,
	storageBucket string,
	storageAuthorityDigest string,
) (storage.ObjectStore, error) {
	provider := strings.ToLower(strings.TrimSpace(storageProvider))
	bucket := strings.TrimSpace(storageBucket)
	if err := validateDurableUploadStorageAuthority(provider, bucket); err != nil {
		return nil, err
	}
	config := r.defaultStorageConfig
	config.Provider = provider
	config.Bucket = bucket
	actualDigest, err := storage.ConfigAuthorityDigest(config)
	if err != nil {
		return nil, fmt.Errorf("%w: %v", errDurableUploadStorageAuthorityInvalid, err)
	}
	if actualDigest != strings.TrimSpace(storageAuthorityDigest) {
		return nil, errDurableUploadStorageAuthorityMismatch
	}
	if provider == strings.ToLower(strings.TrimSpace(r.defaultStorageConfig.Provider)) &&
		bucket == strings.TrimSpace(r.defaultStorageConfig.Bucket) {
		return r.store, nil
	}
	return r.resolveStore(config)
}

func validateDurableUploadStorageAuthority(provider string, bucket string) error {
	switch strings.ToLower(strings.TrimSpace(provider)) {
	case storage.ProviderFilesystem, storage.ProviderS3:
	default:
		return fmt.Errorf(
			"%w: unsupported provider %q",
			errDurableUploadStorageAuthorityInvalid,
			provider,
		)
	}
	if strings.TrimSpace(bucket) == "" {
		return fmt.Errorf(
			"%w: bucket is empty",
			errDurableUploadStorageAuthorityInvalid,
		)
	}
	return nil
}

func validateRuntimeUploadMetadata(
	operation persistence.FileUploadOperation,
	metadata runtimeFileMetadata,
) (string, error) {
	if validateDurableUploadStorageAuthority(
		operation.StorageProvider,
		operation.StorageBucket,
	) != nil ||
		validateDurableUploadStorageAuthority(
			metadata.storageProvider,
			metadata.storageBucket,
		) != nil ||
		metadata.id != operation.SystemFileID || metadata.parentID != operation.ParentID ||
		metadata.parentObject != operation.ParentObjectAPIName ||
		!strings.EqualFold(metadata.storageProvider, operation.StorageProvider) ||
		metadata.storageBucket != operation.StorageBucket ||
		metadata.storagePath != operation.ObjectKey || metadata.storageKey != operation.ObjectKey ||
		metadata.size < 0 || !validSHA256Digest(metadata.checksum) {
		return "", errors.New("runtime file upload lifecycle: metadata identity mismatch")
	}
	logicalObjectKey := systemFileObjectKey(
		metadata.parentObject,
		metadata.parentID,
		metadata.id,
		metadata.originalName,
	)
	if err := validateDurableUploadAttemptIdentity(
		operation,
		logicalObjectKey,
	); err != nil {
		return "", errors.New("runtime file upload lifecycle: attempt object identity mismatch")
	}
	kinds := []string{metadata.attachmentKind}
	if metadata.attachmentKind == defaultRecordAttachmentKind {
		kinds = append(kinds, "")
	}
	matchedHeader := false
	for _, kind := range kinds {
		headerHash, err := durableUploadHeaderHash(contract.FileUploadRequest{
			ID: operation.SystemFileID, FileName: metadata.originalName,
			ContentType: metadata.mimeType, ContentLength: metadata.size,
			ParentID: operation.ParentID, AttachmentKind: kind,
		}, operation.ParentObjectAPIName, logicalObjectKey)
		if err != nil {
			return "", err
		}
		contentDigest, ok := durableUploadContentDigest(operation.RequestHash, headerHash)
		if ok && contentDigest == metadata.checksum {
			matchedHeader = true
			break
		}
	}
	if !matchedHeader {
		return "", errors.New("runtime file upload lifecycle: metadata header mismatch")
	}
	return metadata.checksum, nil
}

func (r *RuntimeUploadLifecycle) completeClaim(ctx context.Context, claim runtimeUploadClaim) error {
	return r.finalizeClaim(ctx, claim, func(tx *sql.Tx) error {
		result, err := tx.ExecContext(ctx, `UPDATE _System_FileUploadOperation
SET State = 'COMPLETED', ErrorCode = NULL, ErrorMessage = NULL,
    ExpiresAt = DATE_ADD(UTC_TIMESTAMP(6), INTERVAL 24 HOUR),
    LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND State = 'FAILED' AND ErrorCode = ? AND ErrorMessage = ?`,
			claim.operation.ID, fileUploadCleanupClaimed, claim.claimToken,
		)
		if err != nil {
			return err
		}
		return requireRuntimeUploadDisposition(result)
	})
}

func (r *RuntimeUploadLifecycle) startPreparedQuietPeriod(
	ctx context.Context,
	claim runtimeUploadClaim,
) error {
	return r.finalizeClaim(ctx, claim, func(tx *sql.Tx) error {
		result, err := tx.ExecContext(ctx, `UPDATE _System_FileUploadOperation
SET ErrorCode = ?, ErrorMessage = NULL,
    ExpiresAt = DATE_ADD(UTC_TIMESTAMP(6), INTERVAL 1 MINUTE),
    LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND State = 'FAILED' AND ErrorCode = ? AND ErrorMessage = ?`,
			fileUploadPreparedCleanupQuiet,
			claim.operation.ID,
			claim.claimCode,
			claim.claimToken,
		)
		if err != nil {
			return err
		}
		return requireRuntimeUploadDisposition(result)
	})
}

func (r *RuntimeUploadLifecycle) deleteClaim(ctx context.Context, claim runtimeUploadClaim) error {
	return r.finalizeClaim(ctx, claim, func(tx *sql.Tx) error {
		result, err := tx.ExecContext(ctx, `DELETE FROM _System_FileUploadOperation
WHERE Id = ? AND State = 'FAILED' AND ErrorCode = ? AND ErrorMessage = ?`,
			claim.operation.ID, claim.claimCode, claim.claimToken,
		)
		if err != nil {
			return err
		}
		return requireRuntimeUploadDisposition(result)
	})
}

func (r *RuntimeUploadLifecycle) poisonClaim(
	ctx context.Context,
	claim runtimeUploadClaim,
	safeCode string,
) error {
	poisonCode := fileUploadRepairPoisoned
	if claim.mode == runtimeUploadClaimPreparedFirstPass ||
		claim.mode == runtimeUploadClaimPreparedFinalPass {
		poisonCode = fileUploadPreparedRepairPoisoned
	}
	return r.finalizeClaim(ctx, claim, func(tx *sql.Tx) error {
		result, err := tx.ExecContext(ctx, `UPDATE _System_FileUploadOperation
SET State = 'FAILED', ErrorCode = ?, ErrorMessage = ?, ExpiresAt = NULL,
    RetryCount = RetryCount + 1,
    LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND State = 'FAILED' AND ErrorCode = ? AND ErrorMessage = ?`,
			poisonCode, safeCode,
			claim.operation.ID, claim.claimCode, claim.claimToken,
		)
		if err != nil {
			return err
		}
		return requireRuntimeUploadDisposition(result)
	})
}

func (r *RuntimeUploadLifecycle) finalizeClaim(
	ctx context.Context,
	claim runtimeUploadClaim,
	disposition func(*sql.Tx) error,
) error {
	tx, cleanup, err := persistence.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return fmt.Errorf("runtime file upload lifecycle: begin disposition: %w", err)
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) error {
		if rollbackErr := tx.Rollback(); rollbackErr != nil {
			return errors.Join(cause, fmt.Errorf("runtime file upload lifecycle: rollback disposition: %w", rollbackErr))
		}
		return cause
	}
	if err := r.ownership.LockActive(ctx, tx); err != nil {
		return rollback(err)
	}
	var ownerToken, requestHash, objectKey, storageProvider, storageBucket string
	var storageAuthorityDigest, state, errorCode, errorMessage string
	err = tx.QueryRowContext(ctx, `SELECT OwnerToken, RequestHash, COALESCE(ObjectKey, ''),
       COALESCE(StorageProvider, ''), COALESCE(StorageBucket, ''),
       COALESCE(StorageAuthorityDigest, ''), State,
       COALESCE(ErrorCode, ''), COALESCE(ErrorMessage, '')
FROM _System_FileUploadOperation WHERE Id = ?
FOR UPDATE`, claim.operation.ID).Scan(
		&ownerToken, &requestHash, &objectKey, &storageProvider, &storageBucket,
		&storageAuthorityDigest,
		&state, &errorCode, &errorMessage,
	)
	if err != nil {
		return rollback(fmt.Errorf("runtime file upload lifecycle: lock disposition: %w", err))
	}
	if ownerToken != claim.operation.OwnerToken || requestHash != claim.operation.RequestHash ||
		objectKey != claim.operation.ObjectKey ||
		storageProvider != claim.operation.StorageProvider ||
		storageBucket != claim.operation.StorageBucket ||
		storageAuthorityDigest != claim.operation.StorageAuthorityDigest ||
		state != persistence.FileUploadOperationFailed ||
		errorCode != claim.claimCode || errorMessage != claim.claimToken {
		return rollback(persistence.ErrFileUploadOperationLostOwnership)
	}
	if err := disposition(tx); err != nil {
		return rollback(err)
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("runtime file upload lifecycle: disposition commit outcome unknown: %w", err)
	}
	return nil
}

func requireRuntimeUploadDisposition(result sql.Result) error {
	affected, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("runtime file upload lifecycle: disposition rows affected: %w", err)
	}
	if affected != 1 {
		return persistence.ErrFileUploadOperationLostOwnership
	}
	return nil
}

func newRuntimeUploadClaimToken() (string, error) {
	var value [16]byte
	if _, err := rand.Read(value[:]); err != nil {
		return "", fmt.Errorf("runtime file upload lifecycle: mint claim token: %w", err)
	}
	return hex.EncodeToString(value[:]), nil
}