package backend

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"time"

	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/meta"
	"go.uber.org/zap"
)

// --- Meta-first saga: upload initiate reservation ---

// reserveUploadOnServer performs the central reserve-first protocol for an
// upload using only the meta DB. AtomicReserveAndInsertUpload claims
// reserved_bytes and inserts the reservation row atomically in the meta DB.
// Either both rows are written or neither is; there is no compensating path that
// can leak reserved_bytes.
//
// Return semantics:
//   - (true, nil):  reservation successfully claimed on the server DB.
//   - (false, ErrStorageQuotaExceeded): quota check failed; tx rolled back
//     so reserved_bytes and the reservation table are untouched.
//   - (false, nil): metaStore is not wired (central quota disabled) OR the
//     server DB is unreachable (fail-open). Callers treat this as "no
//     server-side reservation", so the matching abort/complete paths skip
//     counter adjustments.
//   - (false, nil): central quota stayed lock-contended after bounded retries.
//     This also fail-opens to avoid a user-visible upload-initiate failure, but
//     it records a distinct busy_fail_open metric so quota-admission bypass risk
//     is observable.
//
// Idempotency: a duplicate (tenant_id, upload_id) is treated as
// (true, nil) — the original initiate already claimed the reservation;
// retrying MUST NOT bump reserved_bytes a second time.
func (b *Dat9Backend) reserveUploadOnServer(ctx context.Context, uploadID, targetPath string, totalSize, fileCountDelta int64) (bool, error) {
	if b.metaStore == nil {
		return false, nil
	}
	start := time.Now()

	if _, err := b.metaStore.GetUploadReservation(ctx, b.tenantID, uploadID); err == nil {
		logger.Info(ctx, "central_quota_reserve_upload_duplicate",
			zap.String("tenant_id", b.tenantID),
			zap.String("upload_id", uploadID))
		b.recordTenantOperation("central_quota", "reserve_upload", "duplicate", time.Since(start))
		return true, nil
	} else if err != nil && !errors.Is(err, ErrReservationNotFound) {
		logger.Warn(ctx, "central_quota_reserve_upload_duplicate_lookup_failed",
			zap.String("tenant_id", b.tenantID),
			zap.String("upload_id", uploadID),
			zap.Error(err))
	}

	reservation := &UploadReservationView{
		TenantID:       b.tenantID,
		UploadID:       uploadID,
		ReservedBytes:  totalSize,
		FileCountDelta: fileCountDelta,
		TargetPath:     targetPath,
		Status:         "active",
		ExpiresAt:      time.Now().Add(24 * time.Hour),
	}
	// Hold mutationMu across the pending-delta check and atomic reserve so
	// same-backend post-commit quota mutations cannot slip between them.
	b.mutationMu.Lock()
	defer b.mutationMu.Unlock()
	if err := b.ensureUploadReserveFitsPendingQuota(ctx, totalSize, fileCountDelta); err != nil {
		b.recordTenantOperation("central_quota", "reserve_upload", "quota_exceeded", time.Since(start))
		return false, err
	}
	err := b.metaStore.AtomicReserveAndInsertUpload(ctx, reservation)
	switch {
	case err == nil:
		b.recordTenantOperation("central_quota", "reserve_upload", "ok", time.Since(start))
		return true, nil
	case errors.Is(err, ErrStorageQuotaExceeded):
		b.recordTenantOperation("central_quota", "reserve_upload", "quota_exceeded", time.Since(start))
		return false, err
	case errors.Is(err, ErrFileCountQuotaExceeded):
		b.recordTenantOperation("central_quota", "reserve_upload", "quota_exceeded", time.Since(start))
		return false, err
	case errors.Is(err, ErrReservationAlreadyExists):
		// Idempotent retry: reservation already exists from an earlier initiate.
		// Do NOT bump reserved_bytes again; treat as "server has the reservation".
		logger.Info(ctx, "central_quota_reserve_upload_duplicate",
			zap.String("tenant_id", b.tenantID),
			zap.String("upload_id", uploadID))
		b.recordTenantOperation("central_quota", "reserve_upload", "duplicate", time.Since(start))
		return true, nil
	case errors.Is(err, ErrQuotaReservationBusy):
		logger.Warn(ctx, "central_quota_reserve_upload_busy_fail_open",
			zap.String("tenant_id", b.tenantID),
			zap.String("upload_id", uploadID),
			zap.Int64("size", totalSize),
			zap.Error(err))
		b.recordTenantOperation("central_quota", "reserve_upload", "busy_fail_open", time.Since(start))
		return false, nil
	default:
		logger.Warn(ctx, "central_quota_reserve_upload_failed",
			zap.String("tenant_id", b.tenantID),
			zap.String("upload_id", uploadID),
			zap.Int64("size", totalSize),
			zap.Error(err))
		b.recordTenantOperation("central_quota", "reserve_upload", "fail_open", time.Since(start))
		// Fail-open: allow the upload to proceed when the server DB is down.
		return false, nil
	}
}

func (b *Dat9Backend) ensureUploadReserveFitsPendingQuota(ctx context.Context, totalSize, fileCountDelta int64) error {
	if !b.UseServerQuota() || (totalSize <= 0 && fileCountDelta <= 0) {
		return nil
	}
	cfg := b.cachedQuotaConfig(ctx)
	if cfg == nil {
		b.recordTenantOperation("server_quota", "upload_reserve_pending_check", "fail_open", 0)
		return nil
	}
	if cfg.MaxStorageBytes <= 0 && (fileCountDelta <= 0 || cfg.MaxFileCount <= 0) {
		return nil
	}
	usage := b.loadQuotaUsage(ctx)
	if usage == nil {
		b.recordTenantOperation("server_quota", "upload_reserve_pending_check", "fail_open", 0)
		return nil
	}
	pendingStorageDelta, pendingFileDelta, _ := b.pendingCentralMutationDeltas(ctx)
	projected := usage.StorageBytes + usage.ReservedBytes + pendingStorageDelta + totalSize
	if cfg.MaxStorageBytes > 0 && projected > cfg.MaxStorageBytes {
		return fmt.Errorf("%w: server limit=%d used=%d reserved=%d pending=%d delta=%d",
			ErrStorageQuotaExceeded, cfg.MaxStorageBytes, usage.StorageBytes, usage.ReservedBytes, pendingStorageDelta, totalSize)
	}
	projectedFiles := usage.FileCount + pendingFileDelta + fileCountDelta
	if fileCountDelta > 0 && cfg.MaxFileCount > 0 && projectedFiles > cfg.MaxFileCount {
		return fmt.Errorf("%w: server limit=%d used=%d pending=%d delta=%d",
			ErrFileCountQuotaExceeded, cfg.MaxFileCount, usage.FileCount, pendingFileDelta, fileCountDelta)
	}
	return nil
}

// --- Upload abort: release reservation ---

// abortUploadReservation is called after a tenant-DB upload abort.
// It releases the reserved_bytes back to the pool. Only adjusts counters if a
// server-side reservation row exists and is active/completing, preventing
// counter corruption when the initiate-time reserve was a fail-open no-op.
func (b *Dat9Backend) abortUploadReservation(ctx context.Context, uploadID string, totalSize int64) {
	if b.metaStore == nil {
		return
	}
	start := time.Now()

	aborted := false
	if err := b.metaStore.InTx(ctx, func(tx *sql.Tx) error {
		ok, reservedBytes, fileCountDelta, err := b.metaStore.AbortActiveReservationTx(ctx, tx, b.tenantID, uploadID)
		if err != nil {
			return err
		}
		if !ok {
			return nil
		}
		aborted = true
		if err := b.metaStore.IncrReservedBytesTx(tx, b.tenantID, -reservedBytes); err != nil {
			return err
		}
		if fileCountDelta > 0 {
			if err := b.metaStore.IncrFileCountTx(tx, b.tenantID, -fileCountDelta); err != nil {
				return err
			}
		}
		return nil
	}); err != nil {
		logger.Warn(ctx, "central_quota_abort_failed",
			zap.String("tenant_id", b.tenantID),
			zap.String("upload_id", uploadID),
			zap.Error(err))
		b.recordTenantOperation("central_quota", "abort_reservation", "error", time.Since(start))
		return
	}
	if !aborted {
		b.recordTenantOperation("central_quota", "abort_reservation", "skip_no_reservation", time.Since(start))
		return
	}

	b.recordTenantOperation("central_quota", "abort_reservation", "ok", time.Since(start))
}

// --- Mutation log helpers ---

type uploadCompleteMutationData struct {
	UploadID      string      `json:"upload_id"`
	FileID        string      `json:"file_id"`
	ReservedBytes int64       `json:"reserved_bytes"`
	OldSizeBytes  int64       `json:"old_size_bytes"`
	OldIsMedia    bool        `json:"old_is_media"`
	NewSizeBytes  int64       `json:"new_size_bytes"`
	NewIsMedia    bool        `json:"new_is_media"`
	StorageType   StorageType `json:"storage_type,omitempty"`
}

// completeUploadReservation logs and enqueues the upload-complete saga through
// the same ordered mutation queue as create/overwrite. The reservation-state
// decision ("is this reservation still active? should reserved_bytes be
// transferred?") is made INSIDE the apply/replay transaction via
// SettleActiveReservationTx, not by a pre-check outside the tx. This guarantees
// the durable-outbox contract: the mutation log entry is always written (unless
// InsertMutationLog itself fails), so storage_bytes, file_meta, and media_count
// deltas cannot be silently dropped by a transient lookup error. Any transient
// failure during async apply leaves the mutation in 'pending' status for the
// replay worker.
//
// reservedBytes is the bytes that were claimed at initiate time; the apply tx
// will transfer them reserved→storage only when an active reservation row is
// actually found (settled=true). When no active row exists (fail-open
// initiate or already-completed), the transfer is skipped so counters stay
// consistent.
func (b *Dat9Backend) completeUploadReservation(ctx context.Context, uploadID string, reservedBytes int64, fileID string, oldSizeBytes int64, oldIsMedia bool, newSizeBytes int64, newIsMedia bool) error {
	if b.metaStore == nil {
		return nil
	}
	data := uploadCompleteMutationData{
		UploadID:      uploadID,
		FileID:        fileID,
		ReservedBytes: reservedBytes,
		OldSizeBytes:  oldSizeBytes,
		OldIsMedia:    oldIsMedia,
		NewSizeBytes:  newSizeBytes,
		NewIsMedia:    newIsMedia,
		StorageType:   StorageTypeS3,
	}
	return b.logAndEnqueueMutation(ctx, "upload_complete", data, quotaPendingDeltas{}, func(applyCtx context.Context, tx *sql.Tx) (quotaCounterDeltas, error) {
		return applyUploadCompleteTx(applyCtx, b.metaStore, tx, b.tenantID, data)
	})
}

// applyUploadCompleteTx is the single tx-scoped apply body shared by both the
// backend mutation worker and the replay worker (mutation_replay.go
// upload_complete case). The reservation-state branch lives here (inside the
// tx) so that transient lookup failures outside never cause silent data loss of
// the paired shadow-state mutation.
//
// The settled flag from SettleActiveReservationTx partitions the
// reservation-state space into exactly two branches:
//
//   - settled=true: an 'active' reservation row existed and was atomically
//     transitioned to the terminal status in this tx. The initiate-time
//     claim on reserved_bytes is real, so we transfer reserved → storage.
//
//   - settled=false: no active reservation row found. This covers THREE
//     sub-cases:
//     (1) fail-open initiate — reserveUploadOnServer returned (false, nil)
//     because the server DB was unreachable; no reserved_bytes were
//     ever claimed and no reservation row was written.
//     (2) already-settled terminal row — a concurrent actor (another
//     replay attempt, CLI backfill) completed or aborted the row
//     before we got here.
//     (3) apply-time expiry sweep race — the row existed at initiate time
//     but ExpireActiveReservations released it (and backed out the
//     reserved_bytes) between initiate and this apply tx.
//
//     If file_meta already equals the post-upload state, the row is a
//     duplicate retry after a previous central apply committed but the
//     tenant-local ack did not. In that case this helper is a no-op. Otherwise
//     reserved_bytes is either untouched (case 1) or already balanced
//     (cases 2, 3), so we skip the reserved→storage transfer and charge
//     storage_bytes directly with newSizeBytes.
//
// This function is package-level (not a *Dat9Backend method) so that the
// replay worker — which only holds a MetaQuotaStore, no backend — can call
// it directly without duplicating logic. Caller owns MarkMutationAppliedTx:
// the inline path calls it inside completeUploadReservation's InTx, and the
// replay path calls it inside replayOne. Keeping the mark-applied call out
// of this helper prevents a double-mark rollback trap if any future caller
// also marks applied in the outer tx.
//
// Counter updates are NOT executed inline: the reservation settle and the
// file_meta upsert stay inline, but every tenant_quota_usage adjustment
// (including the reserved_bytes decrement of the reserved→storage transfer)
// is returned as quotaCounterDeltas for the caller to flush at the end of
// the transaction ("hot row last" — see quotaCounterDeltas).
func applyUploadCompleteTx(ctx context.Context, store MetaQuotaStore, tx *sql.Tx, tenantID string, data uploadCompleteMutationData) (quotaCounterDeltas, error) {
	oldExists := false
	var oldMeta *FileMetaView
	if old, err := store.GetFileMetaForUpdateTx(tx, tenantID, data.FileID); err != nil {
		if !errors.Is(err, meta.ErrNotFound) {
			return quotaCounterDeltas{}, err
		}
	} else if old != nil {
		oldExists = true
		oldMeta = old
	}
	oldSizeBytes := data.OldSizeBytes
	oldIsMedia := data.OldIsMedia
	if oldMeta != nil {
		oldSizeBytes = oldMeta.SizeBytes
		oldIsMedia = oldMeta.IsMedia
	}
	mediaDelta := quotaMediaDelta(oldIsMedia, data.NewIsMedia)
	settled, reservedFileCountDelta, err := store.SettleActiveReservationTx(ctx, tx, tenantID, data.UploadID, "completed")
	if err != nil {
		return quotaCounterDeltas{}, err
	}
	newStorageType := storageTypePtr(data.StorageType)
	if !settled && oldMeta != nil && oldMeta.SizeBytes == data.NewSizeBytes && oldMeta.IsMedia == data.NewIsMedia && equalStorageTypes(oldMeta.StorageType, newStorageType) && !oldMeta.LayerOwned {
		return quotaCounterDeltas{}, nil
	}
	var deltas quotaCounterDeltas
	if settled {
		// Transfer reserved → storage as returned counter deltas.
		deltas.reservedBytes -= data.ReservedBytes
		deltas.storageBytes += data.ReservedBytes
	} else {
		// settled=false: fail-open initiate / already-settled / expiry sweep
		// race. reserved_bytes is already balanced; charge storage directly.
		deltas.storageBytes += data.NewSizeBytes
	}
	// A core shadow can alias an immutable LayerFS payload. That payload is
	// accounted by tenant_layer_storage_meta, so replacing the alias must not
	// subtract its bytes or tier from the core counters. The new multipart
	// object owns its bytes directly, regardless of the aliased old state.
	oldStorageBytes := oldSizeBytes
	if oldMeta != nil && oldMeta.LayerOwned {
		oldStorageBytes = 0
	}
	deltas.storageBytes -= oldStorageBytes
	if err := store.UpsertFileMetaTx(tx, &FileMetaView{
		TenantID:    tenantID,
		FileID:      data.FileID,
		SizeBytes:   data.NewSizeBytes,
		IsMedia:     data.NewIsMedia,
		StorageType: newStorageType,
		LayerOwned:  false,
	}); err != nil {
		return quotaCounterDeltas{}, err
	}
	if !oldExists && (!settled || reservedFileCountDelta <= 0) {
		deltas.fileCount = 1
	}
	deltas.mediaFileCount = mediaDelta
	if oldMeta != nil && oldMeta.StorageType != nil && !oldMeta.LayerOwned {
		deltas.addStorageTierBytes(*oldMeta.StorageType, -oldMeta.SizeBytes)
	}
	if newStorageType != nil {
		deltas.addStorageTierBytes(*newStorageType, data.NewSizeBytes)
	}
	return deltas, nil
}