package backend

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

	"github.com/c4pt0r/agfs/agfs-server/pkg/filesystem"
	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/pathutil"
	"github.com/mem9-ai/drive9/pkg/s3client"
	"go.uber.org/zap"
)

// PutFSLayerObject stores a layer file payload outside fs_layer_entries.content_blob.
func (b *Dat9Backend) PutFSLayerObject(ctx context.Context, layerID, path string, body io.Reader, size int64) (*datastore.FSLayerEntry, error) {
	if b == nil {
		return nil, fmt.Errorf("nil backend")
	}
	if b.s3 == nil {
		return nil, fmt.Errorf("s3 client not configured")
	}
	if err := b.ensureUploadSizeAllowed(size); err != nil {
		return nil, err
	}
	if err := b.ensureFileSizeQuota(ctx, size); err != nil {
		return nil, err
	}
	canonical, err := pathutil.Canonicalize(path)
	if err != nil {
		return nil, err
	}
	storageRef := "layers/" + layerID + "/" + b.genID()
	if err := b.prepareFailedWriteObjectGC(ctx, storageRef, ""); err != nil {
		return nil, fmt.Errorf("prepare failed-write object GC: %w", err)
	}
	encOpts, encMode, encKeyID := b.s3WriteEncryption(storageRef)
	hasher := sha256.New()
	sizeLimit := size + 1
	limited := &io.LimitedReader{R: body, N: sizeLimit}
	reader := io.TeeReader(limited, hasher)
	if err := b.s3.PutObject(ctx, storageRef, reader, size, encOpts); err != nil {
		b.failFailedWriteObjectGC(ctx, storageRef, err)
		logger.Error(ctx, "backend_layer_put_object_failed", zap.String("layer_id", layerID), zap.String("path", canonical), zap.String("storage_ref", storageRef), zap.Int64("bytes", size), zap.Error(err))
		return nil, fmt.Errorf("put layer object: %w", err)
	}
	readBytes := sizeLimit - limited.N
	if readBytes == size {
		_, err = io.CopyN(io.Discard, limited, 1)
		if err != nil && !errors.Is(err, io.EOF) {
			b.cleanupFailedWriteObjectGC(ctx, storageRef)
			return nil, fmt.Errorf("verify layer object size: %w", err)
		}
		readBytes = sizeLimit - limited.N
	}
	if readBytes != size {
		b.cleanupFailedWriteObjectGC(ctx, storageRef)
		return nil, fmt.Errorf("layer object size mismatch: read %d bytes, want %d", readBytes, size)
	}
	publishCtx, cancelPublish := context.WithTimeout(ctx, FailedWriteObjectGCPublishTimeout)
	defer cancelPublish()
	if err := b.BeginFailedWriteObjectGCPublish(publishCtx, storageRef); err != nil {
		b.CleanupFailedWriteObjectGC(ctx, storageRef)
		return nil, err
	}
	return &datastore.FSLayerEntry{
		Path:                   canonical,
		Op:                     datastore.FSLayerEntryOpUpsert,
		Kind:                   datastore.FSLayerEntryKindFile,
		StorageType:            string(datastore.StorageS3),
		StorageRef:             storageRef,
		StorageRefHash:         datastore.StorageRefHash(storageRef),
		StorageEncryptionMode:  encMode,
		StorageEncryptionKeyID: encKeyID,
		ChecksumSHA256:         hex.EncodeToString(hasher.Sum(nil)),
		SizeBytes:              size,
	}, nil
}

// InitiateFSLayerUploadV2 creates a direct multipart upload whose completed
// object is published into an active filesystem layer, not the base tree.
func (b *Dat9Backend) InitiateFSLayerUploadV2(ctx context.Context, layerID, path string, totalSize, baseRevision int64, mode uint32) (*UploadPlanV2, error) {
	if b == nil {
		return nil, fmt.Errorf("nil backend")
	}
	if b.s3 == nil {
		return nil, ErrS3NotConfigured
	}
	if totalSize <= 0 {
		return nil, uploadClientProtocolErrorf("total_size must be positive")
	}
	if err := b.ensureUploadSizeAllowed(totalSize); err != nil {
		return nil, err
	}
	if err := b.ensureFileSizeQuota(ctx, totalSize); err != nil {
		return nil, err
	}
	canonical, err := pathutil.Canonicalize(path)
	if err != nil {
		return nil, err
	}
	if err := rejectRootFileNodePath(canonical); err != nil {
		return nil, err
	}
	layer, err := b.store.GetFSLayer(ctx, layerID)
	if err != nil {
		return nil, err
	}
	if layer.State != datastore.FSLayerStateActive {
		return nil, datastore.ErrFSLayerStateConflict
	}
	if !fsLayerBackendPathWithinBase(canonical, layer.BaseRootPath) {
		return nil, fmt.Errorf("fs layer upload path %q is outside base root %q", canonical, layer.BaseRootPath)
	}
	baseInodeID, capturedRevision, err := b.captureFSLayerUploadBase(ctx, canonical, baseRevision)
	if err != nil {
		return nil, err
	}
	if mode == 0 {
		mode = 0o644
	} else {
		mode &= 0o777
	}

	partSize := s3client.CalcAdaptivePartSize(totalSize)
	parts := s3client.CalcParts(totalSize, partSize)
	if len(parts) > MaxMultipartParts {
		return nil, fmt.Errorf("file too large: %d parts exceeds S3 limit of %d", len(parts), MaxMultipartParts)
	}
	objectID := b.genID()
	storageRef := "layers/" + layer.LayerID + "/" + objectID
	encOpts, encMode, encKeyID := b.s3WriteEncryption(storageRef)
	mpu, err := b.s3.CreateMultipartUpload(ctx, storageRef, s3client.ChecksumAlgoNone, encOpts)
	if err != nil {
		return nil, fmt.Errorf("create fs layer multipart upload: %w", err)
	}

	existing, err := b.store.GetFSLayerUploadByPath(ctx, layer.LayerID, canonical)
	if err != nil && !errors.Is(err, datastore.ErrNotFound) {
		_ = b.s3.AbortMultipartUpload(ctx, storageRef, mpu.UploadID)
		return nil, fmt.Errorf("lookup active fs layer upload for %s: %w", canonical, err)
	}
	if existing != nil {
		if err := b.supersedeActiveUpload(ctx, "initiate_fs_layer_upload_v2", existing); err != nil {
			_ = b.s3.AbortMultipartUpload(ctx, storageRef, mpu.UploadID)
			return nil, fmt.Errorf("supersede active fs layer upload for %s: %w", canonical, err)
		}
	}

	now := time.Now().UTC()
	expiresAt := now.Add(24 * time.Hour)
	uploadID := b.genID()
	if err := b.store.InsertUpload(ctx, &datastore.Upload{
		UploadID:               uploadID,
		FileID:                 objectID,
		TargetKind:             datastore.UploadTargetFSLayer,
		LayerID:                layer.LayerID,
		TargetPath:             canonical,
		BaseInodeID:            baseInodeID,
		Mode:                   mode,
		S3UploadID:             mpu.UploadID,
		S3Key:                  storageRef,
		StorageEncryptionMode:  encMode,
		StorageEncryptionKeyID: encKeyID,
		TotalSize:              totalSize,
		PartSize:               partSize,
		PartsTotal:             len(parts),
		ExpectedRevision:       &capturedRevision,
		Status:                 datastore.UploadInitiated,
		CreatedAt:              now,
		UpdatedAt:              now,
		ExpiresAt:              expiresAt,
	}); err != nil {
		_ = b.s3.AbortMultipartUpload(ctx, storageRef, mpu.UploadID)
		return nil, err
	}

	return &UploadPlanV2{
		UploadID:   uploadID,
		Key:        storageRef,
		PartSize:   partSize,
		TotalParts: len(parts),
		ExpiresAt:  expiresAt.Format(time.RFC3339),
		Resumable:  false,
		ChecksumContract: ChecksumContract{
			Supported: []string{"SHA-256"},
			Required:  false,
		},
	}, nil
}

func (b *Dat9Backend) captureFSLayerUploadBase(ctx context.Context, path string, requestedRevision int64) (string, int64, error) {
	if requestedRevision < 0 {
		return "", 0, uploadClientProtocolErrorf("base_revision must be non-negative")
	}
	nf, err := b.store.Stat(ctx, path)
	if errors.Is(err, datastore.ErrNotFound) {
		if requestedRevision > 0 {
			return "", 0, datastore.ErrRevisionConflict
		}
		return "", 0, nil
	}
	if err != nil {
		return "", 0, err
	}
	if nf.Node.IsDirectory || nf.File == nil {
		return "", 0, datastore.ErrPathConflict
	}
	if nf.File.ContentLayout == datastore.ContentLayoutAppendLog {
		return "", 0, ErrAppendLogUnsupported
	}
	if requestedRevision > 0 && requestedRevision != nf.File.Revision {
		return "", 0, datastore.ErrRevisionConflict
	}
	return nf.File.FileID, nf.File.Revision, nil
}

func fsLayerBackendPathWithinBase(path, baseRoot string) bool {
	root := strings.TrimRight(baseRoot, "/")
	return root == "" || path == root || strings.HasPrefix(path, root+"/")
}

// ConfirmFSLayerUploadV2WithChecksum validates and completes a direct layer
// upload, then atomically publishes the resulting immutable S3 object.
func (b *Dat9Backend) ConfirmFSLayerUploadV2WithChecksum(ctx context.Context, uploadID string, clientParts []CompletePart, checksumSHA256 string) (*datastore.FSLayerEntry, error) {
	entry, _, err := b.ConfirmFSLayerUploadV2WithChecksumResult(ctx, uploadID, clientParts, checksumSHA256)
	return entry, err
}

// ConfirmFSLayerUploadV2WithChecksumResult completes a direct layer upload
// and reports whether this call newly published the layer entry.
func (b *Dat9Backend) ConfirmFSLayerUploadV2WithChecksumResult(ctx context.Context, uploadID string, clientParts []CompletePart, checksumSHA256 string) (*datastore.FSLayerEntry, bool, error) {
	if err := validateWholeChecksumSHA256(checksumSHA256); err != nil {
		return nil, false, err
	}
	upload, err := b.store.GetUpload(ctx, uploadID)
	if err != nil {
		return nil, false, err
	}
	if upload.TargetKind != datastore.UploadTargetFSLayer {
		return nil, false, uploadClientProtocolErrorf("upload %s is not an fs layer upload", uploadID)
	}
	entry := fsLayerEntryForUpload(upload, checksumSHA256)
	if upload.Status == datastore.UploadCompleted {
		if err := b.store.PublishFSLayerUpload(ctx, uploadID, entry); err != nil {
			if !errors.Is(err, datastore.ErrCommitOutcomeUnknown) {
				return nil, false, err
			}
			entry, err = b.reconcileFSLayerUploadCommit(ctx, upload, entry)
			if err != nil {
				return nil, false, err
			}
		}
		publishedEntry, err := b.store.GetFSLayerEntryAtSeq(ctx, upload.LayerID, upload.TargetPath, entry.EntrySeq)
		if err != nil {
			return nil, false, err
		}
		b.completeFailedWriteObjectGC(ctx, upload.S3Key)
		if checksumSHA256 != "" && checksumSHA256 != publishedEntry.ChecksumSHA256 {
			return nil, false, fmt.Errorf("%w: completed upload checksum does not match published entry", datastore.ErrUploadConflict)
		}
		if err := b.RecordFSLayerStorage(ctx, publishedEntry); err != nil {
			return nil, false, err
		}
		return publishedEntry, false, nil
	}
	if upload.Status != datastore.UploadUploading {
		return nil, false, datastore.ErrUploadNotActive
	}
	if time.Now().After(upload.ExpiresAt) {
		_ = b.AbortUploadV2(ctx, uploadID)
		return nil, false, datastore.ErrUploadExpired
	}
	s3Parts, _, _, err := b.validateUploadV2Parts(ctx, upload, clientParts)
	if err != nil {
		return nil, false, err
	}
	published, err := b.finalizeFSLayerUpload(ctx, upload, s3Parts, entry)
	if err != nil {
		if !errors.Is(err, datastore.ErrCommitOutcomeUnknown) {
			return nil, false, err
		}
		entry, err = b.reconcileFSLayerUploadCommit(ctx, upload, entry)
		if err != nil {
			b.failFailedWriteObjectGC(ctx, upload.S3Key, err)
			return nil, false, err
		}
		b.completeFailedWriteObjectGC(ctx, upload.S3Key)
		published = true
	}
	if err := b.RecordFSLayerStorage(ctx, entry); err != nil {
		return nil, false, err
	}
	return entry, published, nil
}

func (b *Dat9Backend) reconcileFSLayerUploadCommit(ctx context.Context, upload *datastore.Upload, entry *datastore.FSLayerEntry) (*datastore.FSLayerEntry, error) {
	committed, err := b.store.GetUpload(ctx, upload.UploadID)
	if err != nil {
		return nil, err
	}
	if committed.Status != datastore.UploadCompleted {
		return nil, fmt.Errorf("%w: fs layer upload %s was not committed", datastore.ErrCommitOutcomeUnknown, upload.UploadID)
	}
	resolved, err := b.store.GetFSLayerEntryAtSeq(ctx, upload.LayerID, upload.TargetPath, entry.EntrySeq)
	if err != nil {
		return nil, fmt.Errorf("%w: read committed fs layer upload %s: %v", datastore.ErrCommitOutcomeUnknown, upload.UploadID, err)
	}
	if resolved.StorageRef != upload.S3Key || resolved.SizeBytes != upload.TotalSize || resolved.StorageType != string(datastore.StorageS3) {
		return nil, fmt.Errorf("%w: committed fs layer upload %s metadata mismatch", datastore.ErrCommitOutcomeUnknown, upload.UploadID)
	}
	return resolved, nil
}

func fsLayerEntryForUpload(upload *datastore.Upload, checksumSHA256 string) *datastore.FSLayerEntry {
	baseRevision := int64(0)
	if upload.ExpectedRevision != nil {
		baseRevision = *upload.ExpectedRevision
	}
	return &datastore.FSLayerEntry{
		LayerID:                upload.LayerID,
		Path:                   upload.TargetPath,
		Op:                     datastore.FSLayerEntryOpUpsert,
		Kind:                   datastore.FSLayerEntryKindFile,
		BaseInodeID:            upload.BaseInodeID,
		BaseRevision:           baseRevision,
		StorageType:            string(datastore.StorageS3),
		StorageRef:             upload.S3Key,
		StorageRefHash:         datastore.StorageRefHash(upload.S3Key),
		StorageEncryptionMode:  upload.StorageEncryptionMode,
		StorageEncryptionKeyID: upload.StorageEncryptionKeyID,
		ContentType:            detectContentType(upload.TargetPath, nil),
		ChecksumSHA256:         checksumSHA256,
		SizeBytes:              upload.TotalSize,
		Mode:                   upload.Mode,
	}
}

func (b *Dat9Backend) finalizeFSLayerUpload(ctx context.Context, upload *datastore.Upload, parts []s3client.Part, entry *datastore.FSLayerEntry) (bool, error) {
	publicationCtx, cancelPublication := context.WithTimeout(ctx, FailedWriteObjectGCPublishTimeout)
	defer cancelPublication()
	if err := b.prepareFailedWriteObjectGC(publicationCtx, upload.S3Key, ""); err != nil {
		return false, fmt.Errorf("prepare failed-write object GC: %w", err)
	}
	if err := b.s3.CompleteMultipartUpload(publicationCtx, upload.S3Key, upload.S3UploadID, parts); err != nil {
		b.failFailedWriteObjectGC(ctx, upload.S3Key, err)
		return false, fmt.Errorf("complete fs layer multipart upload: %w", err)
	}

	finalizeCtx, cancelFinalize := postS3UploadFinalizeContext(ctx)
	defer cancelFinalize()
	if err := b.BeginFailedWriteObjectGCPublish(finalizeCtx, upload.S3Key); err != nil {
		b.cleanupFailedFSLayerUpload(finalizeCtx, upload, err)
		return false, err
	}
	published, err := b.store.PublishFSLayerUploadResult(finalizeCtx, upload.UploadID, entry)
	if err != nil {
		if shouldCleanupFailedWriteAfterMetadataError(err) {
			b.cleanupFailedFSLayerUpload(finalizeCtx, upload, err)
		}
		return false, err
	}
	b.completeFailedWriteObjectGC(finalizeCtx, upload.S3Key)
	return published, nil
}

func (b *Dat9Backend) cleanupFailedFSLayerUpload(ctx context.Context, upload *datastore.Upload, publicationErr error) {
	if _, err := b.store.AbortUploadV2(ctx, upload.UploadID); err != nil {
		logger.Warn(ctx, "backend_fs_layer_finalize_abort_metadata_failed", zap.String("upload_id", upload.UploadID), zap.Error(err))
	}
	b.cleanupFailedWriteObjectGC(ctx, upload.S3Key)
	if publicationErr != nil {
		logger.Warn(ctx, "backend_fs_layer_finalize_cleaned_object", zap.String("upload_id", upload.UploadID), zap.Error(publicationErr))
	}
}

// PresignFSLayerEntryData returns a direct read URL for an S3-backed layer entry.
func (b *Dat9Backend) PresignFSLayerEntryData(ctx context.Context, entry *datastore.FSLayerEntry) (string, error) {
	if entry == nil {
		return "", fmt.Errorf("nil fs layer entry")
	}
	if entry.StorageType != string(datastore.StorageS3) || entry.StorageRef == "" {
		return "", ErrNotS3Stored
	}
	if b.s3 == nil {
		return "", ErrS3NotConfigured
	}
	url, err := b.s3.PresignGetObject(ctx, entry.StorageRef, s3client.DownloadTTL)
	if err != nil {
		return "", fmt.Errorf("presign fs layer object %s: %w", entry.StorageRef, err)
	}
	return url, nil
}

// ReadFSLayerEntryData materializes an fs-layer file entry from inline or object storage.
func (b *Dat9Backend) ReadFSLayerEntryData(ctx context.Context, entry *datastore.FSLayerEntry) ([]byte, error) {
	rc, err := b.OpenFSLayerEntryData(ctx, entry)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rc.Close() }()
	return io.ReadAll(rc)
}

// OpenFSLayerEntryData opens an fs-layer file entry for streaming reads.
func (b *Dat9Backend) OpenFSLayerEntryData(ctx context.Context, entry *datastore.FSLayerEntry) (io.ReadCloser, error) {
	if entry == nil {
		return nil, fmt.Errorf("nil fs layer entry")
	}
	if entry.StorageType == string(datastore.StorageS3) || entry.StorageRef != "" {
		if b.s3 == nil {
			return nil, fmt.Errorf("s3 client not configured")
		}
		rc, err := b.s3.GetObject(ctx, entry.StorageRef)
		if err != nil {
			return nil, fmt.Errorf("read layer object %s: %w", entry.StorageRef, err)
		}
		return rc, nil
	}
	return io.NopCloser(bytes.NewReader(entry.ContentBlob)), nil
}

// WriteStoredObjectCtxIfRevision applies an already-stored layer payload to the
// base filesystem without routing the bytes through content_blob.
func (b *Dat9Backend) WriteStoredObjectCtxIfRevision(ctx context.Context, path string, entry *datastore.FSLayerEntry) (int64, error) {
	rev, _, err := b.WriteStoredObjectCreatedCtxIfRevision(ctx, path, entry)
	return rev, err
}

// WriteStoredObjectCreatedCtxIfRevision is WriteStoredObjectCtxIfRevision plus
// inode IDs assigned when the path (and missing parents) are created.
func (b *Dat9Backend) WriteStoredObjectCreatedCtxIfRevision(ctx context.Context, path string, entry *datastore.FSLayerEntry) (rev int64, createdIDs []string, err error) {
	if entry == nil {
		return 0, nil, fmt.Errorf("nil fs layer entry")
	}
	canonical, err := pathutil.Canonicalize(path)
	if err != nil {
		return 0, nil, err
	}
	if err := rejectRootFileNodePath(canonical); err != nil {
		return 0, nil, err
	}
	if entry.StorageRef == "" && (entry.StorageType == "" || entry.StorageType == string(datastore.StorageDB9)) {
		_, rev, createdIDs, err = b.writeCtxIfRevisionWithIdentity(ctx, canonical, entry.ContentBlob, 0, filesystem.WriteFlagCreate|filesystem.WriteFlagTruncate, entry.BaseRevision, strings.TrimSpace(entry.BaseInodeID), nil, "")
		return rev, createdIDs, err
	}
	if entry.StorageType == "" {
		return 0, nil, fmt.Errorf("missing fs layer storage_type for external payload %s", entry.Path)
	}
	if entry.StorageType != string(datastore.StorageS3) {
		return 0, nil, fmt.Errorf("unsupported fs layer storage type %q", entry.StorageType)
	}
	if entry.StorageRef == "" {
		return 0, nil, fmt.Errorf("missing fs layer storage_ref for %s", entry.Path)
	}
	if err := b.ensureUploadSizeAllowed(entry.SizeBytes); err != nil {
		return 0, nil, err
	}
	if err := b.ensureFileSizeQuota(ctx, entry.SizeBytes); err != nil {
		return 0, nil, err
	}
	storageEncryptionMode := entry.StorageEncryptionMode
	if storageEncryptionMode == "" {
		storageEncryptionMode = datastore.StorageEncryptionNone
	}
	storageEncryptionKeyID := entry.StorageEncryptionKeyID

	contentType := detectContentType(canonical, nil)
	now := time.Now()
	nf, statErr := b.store.Stat(ctx, canonical)
	if errorsIsNotFound(statErr) {
		if entry.BaseRevision > 0 || strings.TrimSpace(entry.BaseInodeID) != "" {
			return 0, nil, fmt.Errorf("%w: %w", datastore.ErrRevisionConflict, datastore.ErrFSLayerBasePathMissing)
		}
		fileID := b.genID()
		var createdParents []string
		err := b.store.InTx(ctx, func(tx *sql.Tx) error {
			if err := b.ensureFileCountQuotaServer(ctx, tx, 1); err != nil {
				return err
			}
			if err := b.store.InsertFileTx(tx, &datastore.File{
				FileID:                 fileID,
				StorageType:            datastore.StorageType(entry.StorageType),
				StorageRef:             entry.StorageRef,
				StorageEncryptionMode:  storageEncryptionMode,
				StorageEncryptionKeyID: storageEncryptionKeyID,
				ContentType:            contentType,
				SizeBytes:              entry.SizeBytes,
				ChecksumSHA256:         entry.ChecksumSHA256,
				Revision:               1,
				Status:                 datastore.StatusConfirmed,
				CreatedAt:              now,
				ConfirmedAt:            &now,
			}); err != nil {
				return err
			}
			parents, err := b.store.EnsureParentDirsTxCreated(tx, canonical, b.genID)
			if err != nil {
				return err
			}
			createdParents = parents
			if err := b.store.InsertNodeTx(tx, &datastore.FileNode{
				NodeID:     b.genID(),
				Path:       canonical,
				ParentPath: pathutil.ParentPath(canonical),
				Name:       pathutil.BaseName(canonical),
				FileID:     fileID,
				CreatedAt:  now,
			}); err != nil {
				return err
			}
			return nil
		})
		if err != nil {
			if errors.Is(err, datastore.ErrPathConflict) {
				return 0, nil, fmt.Errorf("%w: %w", datastore.ErrRevisionConflict, datastore.ErrFSLayerBasePathExists)
			}
			return 0, nil, err
		}
		quotaCtx, cancelQuota := postCommitQuotaMutationContext()
		defer cancelQuota()
		createdIDs = append(append([]string{}, createdParents...), fileID)
		if err := b.recordCentralFileCreateMutationWithLayerOwnership(quotaCtx, fileID, entry.SizeBytes, contentType, fsLayerCoreStorageType(datastore.StorageType(entry.StorageType)), entry.StorageType == string(datastore.StorageS3)); err != nil {
			return 1, createdIDs, postCommitQuotaMutationError("record central quota file create", err)
		}
		return 1, createdIDs, nil
	}
	if statErr != nil {
		return 0, nil, statErr
	}
	if nf.File == nil {
		return 0, nil, fmt.Errorf("no file entity")
	}
	if nf.File.ContentLayout == datastore.ContentLayoutAppendLog {
		return 0, nil, ErrAppendLogUnsupported
	}
	if nf.Node.IsDirectory {
		return 0, nil, datastore.ErrPathConflict
	}
	if entry.BaseRevision == 0 && strings.TrimSpace(entry.BaseInodeID) == "" {
		return 0, nil, fmt.Errorf("%w: %w", datastore.ErrRevisionConflict, datastore.ErrFSLayerBasePathExists)
	}

	oldStorageType := nf.File.StorageType
	oldStorageRef := nf.File.StorageRef
	oldSize := nf.File.SizeBytes
	oldContentType := nf.File.ContentType
	var newRev int64
	err = b.store.InTx(ctx, func(tx *sql.Tx) error {
		casFileID := nf.File.FileID
		if strings.TrimSpace(entry.BaseInodeID) != "" {
			live, liveErr := b.store.StatForUpdateTx(ctx, tx, canonical)
			if errors.Is(liveErr, datastore.ErrNotFound) {
				return fmt.Errorf("%w: %w", datastore.ErrRevisionConflict, datastore.ErrFSLayerBasePathMissing)
			}
			if liveErr != nil {
				return liveErr
			}
			if datastore.NodeFileID(live) != entry.BaseInodeID {
				return fmt.Errorf("%w: %w", datastore.ErrRevisionConflict, datastore.ErrFSLayerBaseInodeChanged)
			}
			casFileID = entry.BaseInodeID
		}
		currentMeta, err := b.store.GetFileStorageMetaForUpdateTx(tx, casFileID)
		if err != nil {
			if errors.Is(err, datastore.ErrNotFound) && entry.BaseInodeID != "" {
				return fmt.Errorf("%w: %w", datastore.ErrRevisionConflict, datastore.ErrFSLayerBasePathMissing)
			}
			return err
		}
		if currentMeta.ContentLayout == datastore.ContentLayoutAppendLog {
			return ErrAppendLogUnsupported
		}
		oldStorageType = currentMeta.StorageType
		oldStorageRef = currentMeta.StorageRef
		oldSize = currentMeta.SizeBytes
		oldContentType = currentMeta.ContentType
		if entry.BaseRevision > 0 && currentMeta.Revision != entry.BaseRevision {
			return fmt.Errorf("%w: %w", datastore.ErrRevisionConflict, datastore.ErrFSLayerBaseRevisionChanged)
		}
		var txErr error
		if b.UsesDatabaseAutoEmbedding() {
			if entry.BaseRevision > 0 {
				newRev, txErr = b.store.UpdateFileContentAutoEmbeddingIfRevisionTx(tx, casFileID, entry.BaseRevision, datastore.StorageType(entry.StorageType), entry.StorageRef, contentType, entry.ChecksumSHA256, "", nil, entry.SizeBytes, "")
			} else {
				newRev, txErr = b.store.UpdateFileContentAutoEmbeddingTx(tx, casFileID, datastore.StorageType(entry.StorageType), entry.StorageRef, contentType, entry.ChecksumSHA256, "", nil, entry.SizeBytes, "")
			}
		} else {
			if entry.BaseRevision > 0 {
				newRev, txErr = b.store.UpdateFileContentIfRevisionTx(tx, casFileID, entry.BaseRevision, datastore.StorageType(entry.StorageType), entry.StorageRef, contentType, entry.ChecksumSHA256, "", nil, entry.SizeBytes, "")
			} else {
				newRev, txErr = b.store.UpdateFileContentTx(tx, casFileID, datastore.StorageType(entry.StorageType), entry.StorageRef, contentType, entry.ChecksumSHA256, "", nil, entry.SizeBytes, "")
			}
		}
		if txErr != nil {
			return txErr
		}
		if err := b.store.UpdateFileStorageEncryptionTx(tx, casFileID, storageEncryptionMode, storageEncryptionKeyID); err != nil {
			return err
		}
		return nil
	})
	if err != nil {
		return 0, nil, err
	}
	gcCtx, cancelGC := postCommitQuotaMutationContext()
	b.deleteBlobIfS3Ctx(gcCtx, oldStorageType, oldStorageRef, entry.StorageRef)
	cancelGC()
	quotaCtx, cancelQuota := postCommitQuotaMutationContext()
	defer cancelQuota()
	if err := b.recordCentralFileOverwriteMutationWithLayerOwnership(quotaCtx, nf.File.FileID, oldSize, oldContentType, entry.SizeBytes, contentType, fsLayerCoreStorageType(datastore.StorageType(entry.StorageType)), entry.StorageType == string(datastore.StorageS3)); err != nil {
		return 0, nil, postCommitQuotaMutationError("record central quota file overwrite", err)
	}
	return newRev, nil, nil
}

func errorsIsNotFound(err error) bool {
	return errors.Is(err, datastore.ErrNotFound)
}