package meta

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync/atomic"
	"time"

	mysql "github.com/go-sql-driver/mysql"

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

var errMutationAlreadyApplied = errors.New("mutation already applied")

const (
	defaultMaxMediaLLMFilesValue = int64(100)
	defaultMaxVideoLLMFilesValue = int64(30)
)

func IsMutationAlreadyAppliedError(err error) bool {
	return errors.Is(err, errMutationAlreadyApplied)
}

// QuotaConfig holds per-tenant quota limits stored in the central server DB.
type QuotaConfig struct {
	TenantID               string
	MaxStorageBytes        int64
	MaxFileSizeBytes       int64
	MaxFileCount           int64 // 0 = unlimited
	MaxMediaLLMFiles       int64
	MaxVideoLLMFiles       int64
	MaxMonthlyCostMC       int64 // millicents; 0 = disabled
	TiDBCloudSpendingLimit *int64
	CreatedAt              time.Time
	UpdatedAt              time.Time
}

// QuotaUsage holds pre-aggregated quota counters for a tenant.
type QuotaUsage struct {
	TenantID         string
	StorageBytes     int64
	DBStorageBytes   *int64
	S3StorageBytes   *int64
	StorageTierReady bool
	ReservedBytes    int64
	FileCount        int64
	MediaFileCount   int64
	VideoFileCount   int64
	UpdatedAt        time.Time
}

// TenantQuotaSnapshot groups quota configuration and usage for list responses.
type TenantQuotaSnapshot struct {
	Config QuotaConfig
	Usage  QuotaUsage
}

type StorageType string

const (
	StorageTypeDB9 StorageType = "db9"
	StorageTypeS3  StorageType = "s3"
)

type OrganizationStorageUsage struct {
	TiDBCloudOrgID    string
	StorageBytes      int64
	ConfirmedBytes    int64
	ReservedBytes     int64
	StorageLimitBytes int64
	TenantCount       int64
	FileCount         int64
}

// listOrganizationStorageUsageSQL intentionally includes suspended and deleting
// tenants: their confirmed/reserved bytes are still currently consumed until
// the tenant is released or deleted. It drives from the tenants table so the
// organization tenant count covers every non-deleted tenant, including
// provisioned tenants that do not have a tenant_quota_usage row yet (usage
// rows are created lazily). Unclaimed tenant-pool inventory is excluded the
// same way the control-plane organization visibility predicate excludes it:
// dedicated free bindings via b.pool_status, native-shared free members via
// tenant_pool_memberships. A tenant therefore drops out of both the count and
// the storage sums when its pool claim is released (pool_status='free'), not
// only at deletion. Storage limit keeps its historical semantics: a tenant
// contributes its configured limit only once it has a quota usage row.
const listOrganizationStorageUsageSQL = `SELECT organization_id,
       SUM(storage_bytes + reserved_bytes),
       SUM(storage_bytes),
       SUM(reserved_bytes),
       SUM(storage_limit_bytes),
       COUNT(*),
       SUM(file_count)
FROM (
	SELECT b.organization_id,
	       COALESCE(q.storage_bytes, 0) AS storage_bytes,
	       COALESCE(q.reserved_bytes, 0) AS reserved_bytes,
	       COALESCE(q.file_count, 0) AS file_count,
	       CASE WHEN q.tenant_id IS NULL THEN 0
	            ELSE COALESCE(c.max_storage_bytes, ?) END AS storage_limit_bytes
	FROM tenants t
	STRAIGHT_JOIN tenant_tidbcloud_org_bindings b ON b.tenant_id = t.id
	LEFT JOIN tenant_quota_usage q ON q.tenant_id = t.id
	LEFT JOIN tenant_quota_config c ON c.tenant_id = t.id
	WHERE t.provider <> ? AND t.status <> ? AND b.organization_id <> '' AND b.pool_status <> ?
	UNION ALL
	SELECT d.org_id AS organization_id,
	       COALESCE(q.storage_bytes, 0) AS storage_bytes,
	       COALESCE(q.reserved_bytes, 0) AS reserved_bytes,
	       COALESCE(q.file_count, 0) AS file_count,
	       CASE WHEN q.tenant_id IS NULL THEN 0
	            ELSE COALESCE(c.max_storage_bytes, ?) END AS storage_limit_bytes
	FROM tenants t
	STRAIGHT_JOIN fs_registry f ON f.tenant_id = t.id
	STRAIGHT_JOIN tenant_placements p ON p.fs_id = f.fs_id
	STRAIGHT_JOIN db_pool d ON d.db_id = p.db_id
	LEFT JOIN tenant_pool_memberships m ON m.tenant_id = t.id
	LEFT JOIN tenant_quota_usage q ON q.tenant_id = t.id
	LEFT JOIN tenant_quota_config c ON c.tenant_id = t.id
	WHERE t.provider = ? AND t.status <> ? AND d.org_id <> ''
		AND (m.tenant_id IS NULL OR m.pool_status <> ?)
) organization_usage
GROUP BY organization_id
ORDER BY organization_id`

func (s *Store) ListOrganizationStorageUsage(ctx context.Context) (out []OrganizationStorageUsage, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_organization_storage_usage", start, &err)
	rows, err := s.db.QueryContext(ctx, listOrganizationStorageUsageSQL,
		DefaultMaxStorageBytes(), tidbCloudNativeSharedProvider, TenantDeleted, TenantPoolBindingFree,
		DefaultMaxStorageBytes(), tidbCloudNativeSharedProvider, TenantDeleted, TenantPoolBindingFree,
	)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var usage OrganizationStorageUsage
		if err = rows.Scan(
			&usage.TiDBCloudOrgID,
			&usage.StorageBytes,
			&usage.ConfirmedBytes,
			&usage.ReservedBytes,
			&usage.StorageLimitBytes,
			&usage.TenantCount,
			&usage.FileCount,
		); err != nil {
			return out, err
		}
		out = append(out, usage)
	}
	return out, rows.Err()
}

// QuotaConfigPatch carries externally settable quota limits. Nil fields leave
// existing values unchanged; inserted rows inherit defaults for nil fields.
type QuotaConfigPatch struct {
	MaxStorageBytes        *int64
	MaxFileSizeBytes       *int64
	MaxFileCount           *int64
	MaxMediaLLMFiles       *int64
	MaxVideoLLMFiles       *int64
	TiDBCloudSpendingLimit *int64
}

var ErrTenantExtractConfigNotEnabled = errors.New("enabled tenant extract config is required")

func (p QuotaConfigPatch) AnySet() bool {
	return p.MaxStorageBytes != nil ||
		p.MaxFileSizeBytes != nil ||
		p.MaxFileCount != nil ||
		p.MaxMediaLLMFiles != nil ||
		p.MaxVideoLLMFiles != nil ||
		p.TiDBCloudSpendingLimit != nil
}

// FileMeta tracks per-file quota-relevant metadata in the server DB.
// This is the authoritative source for file size and media classification
// used to compute quota deltas on overwrite/delete.
type FileMeta struct {
	TenantID    string
	FileID      string
	SizeBytes   int64
	IsMedia     bool
	StorageType *StorageType
	LayerOwned  bool
	CreatedAt   time.Time
	UpdatedAt   time.Time
}

// LayerStorageMeta tracks one immutable fs-layer payload version. These rows
// contribute storage only; core file and media counters remain unchanged.
type LayerStorageMeta struct {
	TenantID    string
	LayerID     string
	PathHash    string
	EntrySeq    int64
	SizeBytes   int64
	StorageType StorageType
}

// UploadReservation tracks a single in-flight upload's storage reservation.
type UploadReservation struct {
	TenantID       string
	UploadID       string
	ReservedBytes  int64
	FileCountDelta int64
	TargetPath     string
	Status         string // "active", "completed", "aborted"
	ExpiresAt      time.Time
	CreatedAt      time.Time
	UpdatedAt      time.Time
}

// LLMUsageRecord is a single billable LLM call in the central ledger.
type LLMUsageRecord struct {
	ID             int64
	TenantID       string
	TaskType       string
	TaskID         string
	CostMillicents int64
	RawUnits       int64
	RawUnitType    string
	CreatedAt      time.Time
}

// MutationLogEntry represents a durable outbox entry for quota state mutations.
type MutationLogEntry struct {
	ID             int64
	TenantID       string
	TiDBCloudOrgID string
	MutationType   string // file_create, file_overwrite, file_delete, upload_initiate, upload_complete, upload_abort, llm_cost_record
	MutationData   json.RawMessage
	Status         string // "pending", "applied", "failed"
	RetryCount     int
	CreatedAt      time.Time
	AppliedAt      *time.Time
}

// MutationBacklogObservation is a tenant-level view of pending quota mutation
// replay work for metrics and alerts.
type MutationBacklogObservation struct {
	TenantID                string
	TiDBCloudOrgID          string
	PendingCount            int64
	OldestPendingAgeSeconds float64
}

// --- QuotaConfig operations ---

// GetQuotaConfig returns the per-tenant quota configuration.
// Returns default values if no row exists for the tenant.
func (s *Store) GetQuotaConfig(ctx context.Context, tenantID string) (*QuotaConfig, error) {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "get_quota_config", start, &err)

	cfg := defaultQuotaConfig(tenantID)
	var spendingLimit sql.NullInt64
	err = s.db.QueryRowContext(ctx,
		`SELECT max_storage_bytes, max_file_size_bytes, max_file_count,
		        max_media_llm_files, max_video_llm_files, max_monthly_cost_mc,
		        tidbcloud_spending_limit,
		        created_at, updated_at
		 FROM tenant_quota_config WHERE tenant_id = ?`, tenantID,
	).Scan(&cfg.MaxStorageBytes, &cfg.MaxFileSizeBytes, &cfg.MaxFileCount,
		&cfg.MaxMediaLLMFiles, &cfg.MaxVideoLLMFiles, &cfg.MaxMonthlyCostMC,
		&spendingLimit,
		&cfg.CreatedAt, &cfg.UpdatedAt)
	if err == sql.ErrNoRows {
		return &cfg, nil
	}
	if err != nil {
		return nil, err
	}
	normalizeQuotaConfig(&cfg)
	cfg.TiDBCloudSpendingLimit = ptrFromNullInt64(spendingLimit)
	return &cfg, nil
}

func defaultQuotaConfig(tenantID string) QuotaConfig {
	return QuotaConfig{
		TenantID:         tenantID,
		MaxStorageBytes:  DefaultMaxStorageBytes(),
		MaxFileSizeBytes: DefaultMaxFileSizeBytes(),
		MaxFileCount:     DefaultMaxFileCount(),
		MaxMediaLLMFiles: DefaultMaxMediaLLMFiles(),
		MaxVideoLLMFiles: DefaultMaxVideoLLMFiles(),
	}
}

func normalizeQuotaConfig(cfg *QuotaConfig) {
	if cfg.MaxFileSizeBytes <= 0 {
		cfg.MaxFileSizeBytes = DefaultMaxFileSizeBytes()
	}
}

// SetQuotaConfig upserts per-tenant quota configuration.
func (s *Store) SetQuotaConfig(ctx context.Context, cfg *QuotaConfig) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "set_quota_config", start, &err)

	_, err = s.db.ExecContext(ctx,
		`INSERT INTO tenant_quota_config (tenant_id, max_storage_bytes, max_file_size_bytes, max_file_count,
		                                  max_media_llm_files, max_video_llm_files, max_monthly_cost_mc,
		                                  tidbcloud_spending_limit)
		 VALUES (?, ?, ?, ?, ?, ?, ?, ?)
		 ON DUPLICATE KEY UPDATE
		   max_storage_bytes = VALUES(max_storage_bytes),
		   max_file_size_bytes = VALUES(max_file_size_bytes),
		   max_file_count = VALUES(max_file_count),
		   max_media_llm_files = VALUES(max_media_llm_files),
		   max_video_llm_files = VALUES(max_video_llm_files),
		   max_monthly_cost_mc = VALUES(max_monthly_cost_mc),
		   tidbcloud_spending_limit = COALESCE(VALUES(tidbcloud_spending_limit), tidbcloud_spending_limit)`,
		cfg.TenantID, cfg.MaxStorageBytes, cfg.MaxFileSizeBytes, cfg.MaxFileCount,
		cfg.MaxMediaLLMFiles, cfg.MaxVideoLLMFiles, cfg.MaxMonthlyCostMC,
		nullInt64FromPtr(cfg.TiDBCloudSpendingLimit))
	return err
}

// SetQuotaStorageBytes atomically updates the storage quota, preserving internal
// media/monthly limits when a config row already exists. New rows inherit the
// database defaults for fields that are not externally writable.
func (s *Store) SetQuotaStorageBytes(ctx context.Context, tenantID string, maxStorageBytes int64) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "set_quota_storage_bytes", start, &err)

	_, err = s.db.ExecContext(ctx,
		`INSERT INTO tenant_quota_config (tenant_id, max_storage_bytes, max_file_size_bytes,
		                                  max_media_llm_files, max_video_llm_files)
		 VALUES (?, ?, ?, ?, ?)
		 ON DUPLICATE KEY UPDATE
		   max_storage_bytes = VALUES(max_storage_bytes)`,
		tenantID, maxStorageBytes, DefaultMaxFileSizeBytes(),
		DefaultMaxMediaLLMFiles(), DefaultMaxVideoLLMFiles())
	if err != nil {
		return fmt.Errorf("set quota storage bytes for tenant %q: %w", tenantID, err)
	}
	return nil
}

// SetQuotaConfigPatch updates externally writable quota limits while preserving
// the internal monthly cost limit and any omitted external limits. Explicit LLM
// limits require the matching tenant-specific extract config to be enabled.
func (s *Store) SetQuotaConfigPatch(ctx context.Context, tenantID string, patch QuotaConfigPatch) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "set_quota_config_patch", start, &err)
	err = s.setQuotaConfigPatchWithExtractConfigValidation(ctx, tenantID, patch)
	if err != nil {
		return fmt.Errorf("set quota config patch for tenant %q: %w", tenantID, err)
	}
	return nil
}

// ValidateQuotaExtractConfigs verifies the tenant-specific extract-config
// prerequisites for an external quota patch without changing quota state.
func (s *Store) ValidateQuotaExtractConfigs(ctx context.Context, tenantID string, patch QuotaConfigPatch) error {
	if patch.MaxMediaLLMFiles == nil && patch.MaxVideoLLMFiles == nil {
		return nil
	}
	return s.InTx(ctx, func(tx *sql.Tx) error {
		if err := LockTenantForConfigUpdateTx(ctx, tx, tenantID); err != nil {
			return err
		}
		return s.validateQuotaExtractConfigsTx(ctx, tx, tenantID, patch)
	})
}

func (s *Store) setQuotaConfigPatchWithExtractConfigValidation(ctx context.Context, tenantID string, patch QuotaConfigPatch) error {
	return s.InTx(ctx, func(tx *sql.Tx) error {
		if patch.MaxMediaLLMFiles != nil || patch.MaxVideoLLMFiles != nil {
			if err := LockTenantForConfigUpdateTx(ctx, tx, tenantID); err != nil {
				return err
			}
			if err := s.validateQuotaExtractConfigsTx(ctx, tx, tenantID, patch); err != nil {
				return err
			}
		}
		return upsertTenantQuotaPatchTx(ctx, tx, tenantID, patch)
	})
}

func (s *Store) validateQuotaExtractConfigsTx(ctx context.Context, tx *sql.Tx, tenantID string, patch QuotaConfigPatch) error {
	checks := []struct {
		limit     *int64
		mediaType string
	}{
		{limit: patch.MaxMediaLLMFiles, mediaType: ExtractMediaTypeImage},
		{limit: patch.MaxVideoLLMFiles, mediaType: ExtractMediaTypeVideo},
	}
	for _, check := range checks {
		if check.limit == nil {
			continue
		}
		cfg, err := s.GetTenantExtractConfigForUpdateTx(ctx, tx, tenantID, check.mediaType)
		if errors.Is(err, ErrNotFound) || (err == nil && !cfg.Enabled) {
			return fmt.Errorf("%w for %s", ErrTenantExtractConfigNotEnabled, check.mediaType)
		}
		if err != nil {
			return fmt.Errorf("lookup %s extract config: %w", check.mediaType, err)
		}
	}
	return nil
}

// UpdateSharedTenantQuotaConfig atomically applies the externally writable
// quota fields for a placed shared tenant.
func (s *Store) UpdateSharedTenantQuotaConfig(ctx context.Context, tenantID string, patch QuotaConfigPatch) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "update_shared_tenant_quota_config", start, &err)
	if strings.TrimSpace(tenantID) == "" {
		return fmt.Errorf("tenant id is required")
	}
	if !patch.AnySet() {
		return nil
	}
	return s.InTx(ctx, func(tx *sql.Tx) error {
		if patch.MaxMediaLLMFiles != nil || patch.MaxVideoLLMFiles != nil {
			if err := LockTenantForConfigUpdateTx(ctx, tx, tenantID); err != nil {
				return err
			}
			if err := s.validateQuotaExtractConfigsTx(ctx, tx, tenantID, patch); err != nil {
				return err
			}
		}
		var dbID int64
		if scanErr := tx.QueryRowContext(ctx, `SELECT p.db_id
			FROM tenants t
			JOIN fs_registry f ON f.tenant_id = t.id
			JOIN tenant_placements p ON p.fs_id = f.fs_id
			WHERE t.id = ? AND t.provider = ? AND p.status = ?
			FOR UPDATE`, tenantID, "tidb_cloud_native_shared", SharedDBStatusActive).
			Scan(&dbID); scanErr != nil {
			if errors.Is(scanErr, sql.ErrNoRows) {
				return ErrNotFound
			}
			return fmt.Errorf("lock shared tenant placement: %w", scanErr)
		}
		var maxStorage, maxFileSize, maxFileCount, maxMediaLLM, maxVideoLLM int64
		if scanErr := tx.QueryRowContext(ctx, `SELECT max_storage_bytes, max_file_size_bytes,
			max_file_count, max_media_llm_files, max_video_llm_files
			FROM tenant_quota_config WHERE tenant_id = ? FOR UPDATE`, tenantID).
			Scan(&maxStorage, &maxFileSize, &maxFileCount, &maxMediaLLM, &maxVideoLLM); scanErr != nil {
			if errors.Is(scanErr, sql.ErrNoRows) {
				return ErrSharedDBQuotaNotMaterialized
			}
			return fmt.Errorf("lock shared tenant quota: %w", scanErr)
		}
		if patch.MaxStorageBytes != nil {
			maxStorage = *patch.MaxStorageBytes
		}
		if patch.MaxFileSizeBytes != nil {
			maxFileSize = *patch.MaxFileSizeBytes
		}
		if patch.MaxFileCount != nil {
			maxFileCount = *patch.MaxFileCount
		}
		if patch.MaxMediaLLMFiles != nil {
			maxMediaLLM = *patch.MaxMediaLLMFiles
		}
		if patch.MaxVideoLLMFiles != nil {
			maxVideoLLM = *patch.MaxVideoLLMFiles
		}
		res, updateErr := tx.ExecContext(ctx, `UPDATE tenant_quota_config
			SET max_storage_bytes = ?, max_file_size_bytes = ?, max_file_count = ?,
				max_media_llm_files = ?, max_video_llm_files = ?
			WHERE tenant_id = ?`, maxStorage, maxFileSize, maxFileCount, maxMediaLLM, maxVideoLLM, tenantID)
		if updateErr != nil {
			return fmt.Errorf("update shared tenant quota: %w", updateErr)
		}
		affected, rowsErr := res.RowsAffected()
		if rowsErr != nil {
			return rowsErr
		}
		if affected == 0 {
			return ErrNotFound
		}
		return nil
	})
}

func ptrFromNullInt64(v sql.NullInt64) *int64 {
	if !v.Valid {
		return nil
	}
	out := v.Int64
	return &out
}

func nullTimeFromPtr(v *time.Time) sql.NullTime {
	if v == nil {
		return sql.NullTime{}
	}
	// DATETIME columns are timezone-less; normalize to UTC before bind. Paired
	// with mysqlutil.NormalizeMetaDSNTimeContract on RoleMeta opens, this keeps
	// stored wall clocks comparable to UTC_TIMESTAMP(3).
	return sql.NullTime{Time: v.UTC(), Valid: true}
}

func ptrFromNullTime(v sql.NullTime) *time.Time {
	if !v.Valid {
		return nil
	}
	out := v.Time
	return &out
}

func (s *Store) CopyQuotaConfig(ctx context.Context, sourceTenantID, destTenantID string) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "copy_quota_config", start, &err)

	cfg, err := s.GetQuotaConfig(ctx, sourceTenantID)
	if err != nil {
		return err
	}
	cfg.TenantID = destTenantID
	return s.SetQuotaConfig(ctx, cfg)
}

// --- QuotaUsage operations ---

// TierFileCounts is a low-frequency observation of a tenant's confirmed file
// count and its storage-tier split. FileCount is derived from
// tenant_file_meta (one row per confirmed file), so it never includes
// file_count reservations claimed by in-flight uploads. DBFileCount/
// S3FileCount are nil unless TierReady is true (the same readiness gate as
// the storage-bytes split).
type TierFileCounts struct {
	FileCount   int64
	DBFileCount *int64
	S3FileCount *int64
	TierReady   bool
}

// GetTierFileCounts returns the tenant's confirmed file count plus, when the
// storage tier scan is ready, the per-tier split. All counts come from
// tenant_file_meta: layer-owned aliases are files (their bytes are not) and
// rows created by upload reservations only appear at completion, so the
// result is a true confirmed count. It exists for background metric refresh
// only — the COUNT reads must never run on request hot paths.
func (s *Store) GetTierFileCounts(ctx context.Context, tenantID string) (out *TierFileCounts, err error) {
	start := time.Now()
	defer observeMeta(ctx, "get_tier_file_counts", start, &err)

	out = &TierFileCounts{}
	if err = s.db.QueryRowContext(ctx,
		`SELECT COUNT(*) FROM tenant_file_meta WHERE tenant_id = ?`, tenantID,
	).Scan(&out.FileCount); err != nil {
		return nil, err
	}
	// Tier readiness gate: same contract as the storage-bytes split in
	// GetQuotaUsage. A missing quota row simply means not ready.
	if err = s.db.QueryRowContext(ctx,
		`SELECT COALESCE((
		     SELECT CASE WHEN s.status = 'completed' AND s.full_scan_completed_at IS NOT NULL
		                      AND q.db_storage_bytes IS NOT NULL AND q.s3_storage_bytes IS NOT NULL
		                      AND q.db_storage_bytes >= 0 AND q.s3_storage_bytes >= 0
		                      AND q.db_storage_bytes + q.s3_storage_bytes = q.storage_bytes
		                      AND NOT EXISTS (
		                          SELECT 1 FROM tenant_file_meta f
		                          WHERE f.tenant_id = q.tenant_id AND f.storage_type IS NULL
		                      )
		                      AND NOT EXISTS (
		                          SELECT 1 FROM tenant_layer_storage_meta l
		                          WHERE l.tenant_id = q.tenant_id AND l.storage_type IS NULL
		                      )
		                 THEN 1 ELSE 0 END
		     FROM tenant_quota_usage q
		     LEFT JOIN tenant_storage_tier_scan s ON s.tenant_id = q.tenant_id
		     WHERE q.tenant_id = ?), 0)`, tenantID,
	).Scan(&out.TierReady); err != nil {
		return nil, err
	}
	if !out.TierReady {
		return out, nil
	}
	rows, err := s.db.QueryContext(ctx,
		`SELECT storage_type, COUNT(*) FROM tenant_file_meta WHERE tenant_id = ? GROUP BY storage_type`, tenantID)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	dbFileCount, s3FileCount := int64(0), int64(0)
	for rows.Next() {
		var storageType string
		var count int64
		if err := rows.Scan(&storageType, &count); err != nil {
			return nil, err
		}
		switch storageType {
		case string(StorageTypeDB9):
			dbFileCount = count
		case string(StorageTypeS3):
			s3FileCount = count
		default:
			// A future/unknown tier disables the split, mirroring the
			// storage-bytes contract for unclassified rows.
			out.TierReady = false
			return out, rows.Err()
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	out.DBFileCount = &dbFileCount
	out.S3FileCount = &s3FileCount
	return out, nil
}

// GetQuotaUsage returns the pre-aggregated quota counters for a tenant.
func (s *Store) GetQuotaUsage(ctx context.Context, tenantID string) (*QuotaUsage, error) {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "get_quota_usage", start, &err)

	u := &QuotaUsage{TenantID: tenantID}
	var dbStorageBytes, s3StorageBytes sql.NullInt64
	err = s.db.QueryRowContext(ctx,
		`SELECT q.storage_bytes, q.db_storage_bytes, q.s3_storage_bytes,
		        q.reserved_bytes, q.file_count, q.media_file_count, q.video_file_count, q.updated_at,
		        CASE WHEN s.status = 'completed' AND s.full_scan_completed_at IS NOT NULL
		                  AND q.db_storage_bytes IS NOT NULL AND q.s3_storage_bytes IS NOT NULL
		                  AND q.db_storage_bytes >= 0 AND q.s3_storage_bytes >= 0
		                  AND q.db_storage_bytes + q.s3_storage_bytes = q.storage_bytes
		                  AND NOT EXISTS (
		                      SELECT 1 FROM tenant_file_meta f
		                      WHERE f.tenant_id = q.tenant_id AND f.storage_type IS NULL
		                  )
		                  AND NOT EXISTS (
		                      SELECT 1 FROM tenant_layer_storage_meta l
		                      WHERE l.tenant_id = q.tenant_id AND l.storage_type IS NULL
		                  )
		             THEN 1 ELSE 0 END
		 FROM tenant_quota_usage q
		 LEFT JOIN tenant_storage_tier_scan s ON s.tenant_id = q.tenant_id
		 WHERE q.tenant_id = ?`, tenantID,
	).Scan(&u.StorageBytes, &dbStorageBytes, &s3StorageBytes, &u.ReservedBytes,
		&u.FileCount, &u.MediaFileCount, &u.VideoFileCount, &u.UpdatedAt, &u.StorageTierReady)
	if err == sql.ErrNoRows {
		return u, nil // zero counters
	}
	if err != nil {
		return nil, err
	}
	u.DBStorageBytes = ptrFromNullInt64(dbStorageBytes)
	u.S3StorageBytes = ptrFromNullInt64(s3StorageBytes)
	return u, nil
}

// ListTenantQuotaSnapshots returns quota configuration and usage for the
// requested tenants using two queries. Missing rows use the same defaults as
// GetQuotaConfig and GetQuotaUsage.
func (s *Store) ListTenantQuotaSnapshots(ctx context.Context, tenantIDs []string) (out map[string]TenantQuotaSnapshot, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_tenant_quota_snapshots", start, &err)

	out = make(map[string]TenantQuotaSnapshot, len(tenantIDs))
	if len(tenantIDs) == 0 {
		return out, nil
	}
	placeholders := make([]string, len(tenantIDs))
	args := make([]any, len(tenantIDs))
	for i, tenantID := range tenantIDs {
		placeholders[i] = "?"
		args[i] = tenantID
		out[tenantID] = TenantQuotaSnapshot{
			Config: defaultQuotaConfig(tenantID),
			Usage:  QuotaUsage{TenantID: tenantID},
		}
	}

	rows, err := s.db.QueryContext(ctx,
		`SELECT tenant_id, max_storage_bytes, max_file_size_bytes, max_file_count,
		        max_media_llm_files, max_video_llm_files, max_monthly_cost_mc,
		        tidbcloud_spending_limit,
		        created_at, updated_at
		 FROM tenant_quota_config WHERE tenant_id IN (`+strings.Join(placeholders, ",")+`)`, args...)
	if err != nil {
		return nil, err
	}
	for rows.Next() {
		var cfg QuotaConfig
		var spendingLimit sql.NullInt64
		if err = rows.Scan(&cfg.TenantID, &cfg.MaxStorageBytes, &cfg.MaxFileSizeBytes, &cfg.MaxFileCount,
			&cfg.MaxMediaLLMFiles, &cfg.MaxVideoLLMFiles, &cfg.MaxMonthlyCostMC,
			&spendingLimit,
			&cfg.CreatedAt, &cfg.UpdatedAt); err != nil {
			_ = rows.Close()
			return nil, err
		}
		normalizeQuotaConfig(&cfg)
		cfg.TiDBCloudSpendingLimit = ptrFromNullInt64(spendingLimit)
		snapshot := out[cfg.TenantID]
		snapshot.Config = cfg
		out[cfg.TenantID] = snapshot
	}
	if err = rows.Err(); err != nil {
		_ = rows.Close()
		return nil, err
	}
	if err = rows.Close(); err != nil {
		return nil, err
	}

	rows, err = s.db.QueryContext(ctx,
		`SELECT tenant_id, storage_bytes, reserved_bytes, file_count, media_file_count, video_file_count, updated_at
		 FROM tenant_quota_usage WHERE tenant_id IN (`+strings.Join(placeholders, ",")+`)`, args...)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var usage QuotaUsage
		if err = rows.Scan(&usage.TenantID, &usage.StorageBytes, &usage.ReservedBytes, &usage.FileCount,
			&usage.MediaFileCount, &usage.VideoFileCount, &usage.UpdatedAt); err != nil {
			return nil, err
		}
		snapshot := out[usage.TenantID]
		snapshot.Usage = usage
		out[usage.TenantID] = snapshot
	}
	return out, rows.Err()
}

// IncrFileCount atomically adjusts the file_count counter.
func (s *Store) IncrFileCount(ctx context.Context, tenantID string, delta int64) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "incr_file_count", start, &err)

	res, err := s.db.ExecContext(ctx,
		`UPDATE tenant_quota_usage SET file_count = file_count + ? WHERE tenant_id = ?`,
		delta, tenantID)
	if err != nil {
		return err
	}
	return ensureRowsAffected(res, tenantID)
}

// IncrFileCountTx atomically adjusts the file_count counter inside a transaction.
func (s *Store) IncrFileCountTx(tx *sql.Tx, tenantID string, delta int64) error {
	res, err := tx.Exec(
		`UPDATE tenant_quota_usage SET file_count = file_count + ? WHERE tenant_id = ?`,
		delta, tenantID)
	if err != nil {
		return err
	}
	return ensureRowsAffected(res, tenantID)
}

// EnsureQuotaUsageRow creates a zero-valued quota usage row if none exists.
func (s *Store) EnsureQuotaUsageRow(ctx context.Context, tenantID string) error {
	_, err := s.db.ExecContext(ctx,
		`INSERT IGNORE INTO tenant_quota_usage (tenant_id) VALUES (?)`, tenantID)
	return err
}

// ensureRowsAffected checks that an UPDATE touched at least one row.
// Returns an error if the tenant_quota_usage row is missing.
func ensureRowsAffected(res sql.Result, tenantID string) error {
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n == 0 {
		return fmt.Errorf("tenant_quota_usage row missing for tenant %s", tenantID)
	}
	return nil
}

// IncrStorageBytes atomically adjusts the storage_bytes counter.
func (s *Store) IncrStorageBytes(ctx context.Context, tenantID string, delta int64) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "incr_storage_bytes", start, &err)

	res, err := s.db.ExecContext(ctx,
		`UPDATE tenant_quota_usage SET storage_bytes = storage_bytes + ? WHERE tenant_id = ?`,
		delta, tenantID)
	if err != nil {
		return err
	}
	return ensureRowsAffected(res, tenantID)
}

// IncrStorageBytesTx atomically adjusts the storage_bytes counter inside a transaction.
func (s *Store) IncrStorageBytesTx(tx *sql.Tx, tenantID string, delta int64) error {
	res, err := tx.Exec(
		`UPDATE tenant_quota_usage SET storage_bytes = storage_bytes + ? WHERE tenant_id = ?`,
		delta, tenantID)
	if err != nil {
		return err
	}
	return ensureRowsAffected(res, tenantID)
}

// IncrReservedBytes atomically adjusts the reserved_bytes counter.
func (s *Store) IncrReservedBytes(ctx context.Context, tenantID string, delta int64) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "incr_reserved_bytes", start, &err)

	res, err := s.db.ExecContext(ctx,
		`UPDATE tenant_quota_usage SET reserved_bytes = reserved_bytes + ? WHERE tenant_id = ?`,
		delta, tenantID)
	if err != nil {
		return err
	}
	return ensureRowsAffected(res, tenantID)
}

// IncrReservedBytesTx atomically adjusts the reserved_bytes counter inside a transaction.
func (s *Store) IncrReservedBytesTx(tx *sql.Tx, tenantID string, delta int64) error {
	res, err := tx.Exec(
		`UPDATE tenant_quota_usage SET reserved_bytes = reserved_bytes + ? WHERE tenant_id = ?`,
		delta, tenantID)
	if err != nil {
		return err
	}
	return ensureRowsAffected(res, tenantID)
}

// IncrMediaFileCount atomically adjusts the media_file_count counter.
func (s *Store) IncrMediaFileCount(ctx context.Context, tenantID string, delta int64) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "incr_media_file_count", start, &err)

	res, err := s.db.ExecContext(ctx,
		`UPDATE tenant_quota_usage SET media_file_count = media_file_count + ? WHERE tenant_id = ?`,
		delta, tenantID)
	if err != nil {
		return err
	}
	return ensureRowsAffected(res, tenantID)
}

// IncrMediaFileCountTx atomically adjusts the media_file_count counter inside a transaction.
func (s *Store) IncrMediaFileCountTx(tx *sql.Tx, tenantID string, delta int64) error {
	res, err := tx.Exec(
		`UPDATE tenant_quota_usage SET media_file_count = media_file_count + ? WHERE tenant_id = ?`,
		delta, tenantID)
	if err != nil {
		return err
	}
	return ensureRowsAffected(res, tenantID)
}

// IncrVideoFileCount atomically adjusts the video_file_count counter.
func (s *Store) IncrVideoFileCount(ctx context.Context, tenantID string, delta int64) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "incr_video_file_count", start, &err)

	res, err := s.db.ExecContext(ctx,
		`UPDATE tenant_quota_usage SET video_file_count = video_file_count + ? WHERE tenant_id = ?`,
		delta, tenantID)
	if err != nil {
		return err
	}
	return ensureRowsAffected(res, tenantID)
}

// IncrVideoFileCountTx atomically adjusts the video_file_count counter inside a transaction.
func (s *Store) IncrVideoFileCountTx(tx *sql.Tx, tenantID string, delta int64) error {
	res, err := tx.Exec(
		`UPDATE tenant_quota_usage SET video_file_count = video_file_count + ? WHERE tenant_id = ?`,
		delta, tenantID)
	if err != nil {
		return err
	}
	return ensureRowsAffected(res, tenantID)
}

// TransferReservedToConfirmed atomically moves bytes from reserved to confirmed storage.
func (s *Store) TransferReservedToConfirmed(ctx context.Context, tenantID string, reservedDelta, storageDelta int64) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "transfer_reserved_to_confirmed", start, &err)

	res, err := s.db.ExecContext(ctx,
		`UPDATE tenant_quota_usage
		 SET reserved_bytes = reserved_bytes + ?, storage_bytes = storage_bytes + ?
		 WHERE tenant_id = ?`,
		reservedDelta, storageDelta, tenantID)
	if err != nil {
		return err
	}
	return ensureRowsAffected(res, tenantID)
}

// TransferReservedToConfirmedTx atomically moves bytes from reserved to confirmed storage in a transaction.
func (s *Store) TransferReservedToConfirmedTx(tx *sql.Tx, tenantID string, reservedDelta, storageDelta int64) error {
	res, err := tx.Exec(
		`UPDATE tenant_quota_usage
		 SET reserved_bytes = reserved_bytes + ?, storage_bytes = storage_bytes + ?
		 WHERE tenant_id = ?`,
		reservedDelta, storageDelta, tenantID)
	if err != nil {
		return err
	}
	return ensureRowsAffected(res, tenantID)
}

// IncrQuotaUsageCountersTx atomically adjusts all quota usage counters
// with a single UPDATE inside a transaction. It mirrors the per-counter
// Incr*Tx semantics exactly (plain additive deltas, no clamping, and the
// tenant row must already exist), but collapses what used to be one
// statement per touched counter into one statement for the whole batch.
// The batched mutation apply paths call this at the END of their
// transaction ("hot row last") so the per-tenant tenant_quota_usage row is
// locked only just before commit instead of from the first mutation apply.
// It is a no-op when every delta is zero, so callers may invoke it
// unconditionally.
func (s *Store) IncrQuotaUsageCountersTx(tx *sql.Tx, tenantID string, storageDelta, dbStorageDelta, s3StorageDelta, fileDelta, mediaDelta, reservedDelta int64) error {
	if storageDelta == 0 && dbStorageDelta == 0 && s3StorageDelta == 0 && fileDelta == 0 && mediaDelta == 0 && reservedDelta == 0 {
		return nil
	}
	res, err := tx.Exec(
		`UPDATE tenant_quota_usage
		 SET storage_bytes = storage_bytes + ?,
		     db_storage_bytes = db_storage_bytes + ?,
		     s3_storage_bytes = s3_storage_bytes + ?,
		     file_count = file_count + ?,
		     media_file_count = media_file_count + ?,
		     reserved_bytes = reserved_bytes + ?
		 WHERE tenant_id = ?`,
		storageDelta, dbStorageDelta, s3StorageDelta, fileDelta, mediaDelta, reservedDelta, tenantID)
	if err != nil {
		return err
	}
	return ensureRowsAffected(res, tenantID)
}

// defaultMaxStorageBytes is the fallback limit when no per-tenant config row exists.
var defaultMaxStorageBytes atomic.Int64
var defaultMaxFileSizeBytes atomic.Int64
var defaultMaxFileCount atomic.Int64
var defaultMaxMediaLLMFiles atomic.Int64
var defaultMaxVideoLLMFiles atomic.Int64

func init() {
	defaultMaxStorageBytes.Store(int64(50 * (1 << 30)))  // 50 GiB
	defaultMaxFileSizeBytes.Store(int64(10 * (1 << 30))) // 10 GiB
	defaultMaxFileCount.Store(0)                         // unlimited
	defaultMaxMediaLLMFiles.Store(defaultMaxMediaLLMFilesValue)
	defaultMaxVideoLLMFiles.Store(defaultMaxVideoLLMFilesValue)
}

// SetDefaultMaxStorageBytes overrides the per-tenant fallback storage quota.
func SetDefaultMaxStorageBytes(bytes int64) {
	if bytes > 0 {
		defaultMaxStorageBytes.Store(bytes)
	}
}

// DefaultMaxStorageBytes returns the configured per-tenant fallback storage quota.
func DefaultMaxStorageBytes() int64 { return defaultMaxStorageBytes.Load() }

// SetDefaultMaxFileSizeBytes overrides the per-tenant fallback file size quota.
func SetDefaultMaxFileSizeBytes(bytes int64) {
	if bytes > 0 {
		defaultMaxFileSizeBytes.Store(bytes)
	}
}

// DefaultMaxFileSizeBytes returns the configured per-tenant fallback file size quota.
func DefaultMaxFileSizeBytes() int64 { return defaultMaxFileSizeBytes.Load() }

// SetDefaultMaxFileCount overrides the per-tenant fallback file count quota.
// Zero means unlimited. Negative values are rejected by leaving the current
// value unchanged.
func SetDefaultMaxFileCount(count int64) {
	if count >= 0 {
		defaultMaxFileCount.Store(count)
	}
}

// DefaultMaxFileCount returns the configured per-tenant fallback file count.
func DefaultMaxFileCount() int64 { return defaultMaxFileCount.Load() }

// SetDefaultMaxMediaLLMFiles overrides the per-tenant fallback media LLM file limit.
func SetDefaultMaxMediaLLMFiles(n int64) {
	if n > 0 {
		defaultMaxMediaLLMFiles.Store(n)
	}
}

// DefaultMaxMediaLLMFiles returns the configured per-tenant fallback media LLM file limit.
func DefaultMaxMediaLLMFiles() int64 { return defaultMaxMediaLLMFiles.Load() }

// SetDefaultMaxVideoLLMFiles overrides the per-tenant fallback video LLM file limit.
func SetDefaultMaxVideoLLMFiles(n int64) {
	if n > 0 {
		defaultMaxVideoLLMFiles.Store(n)
	}
}

// DefaultMaxVideoLLMFiles returns the configured per-tenant fallback video LLM file limit.
func DefaultMaxVideoLLMFiles() int64 { return defaultMaxVideoLLMFiles.Load() }

const (
	metaLockConflictRetryAttempts = 4
	metaLockConflictRetryBase     = 20 * time.Millisecond
)

// AtomicReserveAndInsertUpload claims reserved_bytes and inserts the reservation
// tracking row inside a single server DB transaction. This is the correct API
// for the upload-initiate path: either both rows are written, or neither is.
//
// Returns:
//   - ErrStorageQuotaExceeded when the projected storage total exceeds the
//     per-tenant limit (or the default when no config row exists). Transaction
//     rolls back so reserved_bytes and the reservation table are both untouched.
//   - ErrFileCountQuotaExceeded when the projected file count exceeds the
//     per-tenant limit. Transaction rolls back.
//   - ErrReservationAlreadyExists when a row with the same (tenant_id, upload_id)
//     already exists (primary-key collision). Idempotent-retry callers should
//     detect this sentinel and NOT bump reserved_bytes a second time.
//   - Any other DB error causes the transaction to roll back.
func (s *Store) AtomicReserveAndInsertUpload(ctx context.Context, r *UploadReservation) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "atomic_reserve_and_insert_upload", start, &err)

	err = retryMetaLockConflict(ctx, r.TenantID, "reserve_upload", "atomic_reserve_and_insert_upload", func() error {
		return s.atomicReserveAndInsertUploadOnce(ctx, r)
	})
	return err
}

func (s *Store) atomicReserveAndInsertUploadOnce(ctx context.Context, r *UploadReservation) error {
	return s.InTx(ctx, func(tx *sql.Tx) error {
		if err := s.claimOrBootstrapUploadReservationQuotaTx(ctx, tx, r); err != nil {
			return err
		}
		if _, execErr := tx.ExecContext(ctx,
			`INSERT INTO tenant_upload_reservations
			   (tenant_id, upload_id, reserved_bytes, file_count_delta, target_path, status, expires_at)
			 VALUES (?, ?, ?, ?, ?, 'active', ?)`,
			r.TenantID, r.UploadID, r.ReservedBytes, r.FileCountDelta, r.TargetPath, r.ExpiresAt); execErr != nil {
			if isDuplicateEntry(execErr) {
				return ErrReservationAlreadyExists
			}
			return execErr
		}
		return nil
	})
}

// claimOrBootstrapUploadReservationQuotaTx claims reserved_bytes / file_count
// with an UPDATE-first path. INSERT IGNORE is not used on the hot path: on
// an existing PRIMARY record it takes a shared lock, and two concurrent
// reserves then deadlock on the S→X upgrade of the same tenant_quota_usage
// row. A missing usage row (new tenant) is inserted only after the UPDATE
// reports 0 rows and a consistent-read SELECT confirms the row is absent.
func (s *Store) claimOrBootstrapUploadReservationQuotaTx(ctx context.Context, tx *sql.Tx, r *UploadReservation) error {
	claimed, err := s.claimUploadReservationQuotaTx(ctx, tx, r)
	if err != nil {
		return err
	}
	if claimed {
		return nil
	}
	quotaErr := s.uploadReservationQuotaErrorTx(ctx, tx, r)
	if errors.Is(quotaErr, errQuotaUsageRowMissing) {
		if insertErr := s.insertQuotaUsageRowTx(ctx, tx, r.TenantID); insertErr != nil {
			return insertErr
		}
	} else if quotaErr != nil {
		return quotaErr
	}
	// Retry the claim after a missing-row bootstrap, or when attribution
	// was nil: a peer released reserved bytes between the current-read
	// UPDATE and the first consistent read, or (READ COMMITTED / TiDB) a
	// peer created the usage row in that window. INSERT is skipped when
	// the row already exists so the hot-row S lock is never taken on the
	// existing-tenant path.
	claimed, err = s.claimUploadReservationQuotaTx(ctx, tx, r)
	if err != nil {
		return err
	}
	if claimed {
		return nil
	}
	quotaErr = s.uploadReservationQuotaErrorTx(ctx, tx, r)
	if quotaErr == nil || errors.Is(quotaErr, errQuotaUsageRowMissing) {
		return ErrStorageQuotaExceeded
	}
	return quotaErr
}

func (s *Store) claimUploadReservationQuotaTx(ctx context.Context, tx *sql.Tx, r *UploadReservation) (bool, error) {
	res, err := tx.ExecContext(ctx,
		`UPDATE tenant_quota_usage
		 SET reserved_bytes = reserved_bytes + ?,
		     file_count = file_count + ?
		 WHERE tenant_id = ?
		   AND storage_bytes + reserved_bytes + ? <=
		       COALESCE((SELECT max_storage_bytes
		                 FROM tenant_quota_config WHERE tenant_id = ?), ?)
		   AND (? <= 0 OR
		        COALESCE((SELECT max_file_count
		                  FROM tenant_quota_config WHERE tenant_id = ?), 0) <= 0 OR
		        file_count + ? <= COALESCE((SELECT max_file_count
		                                    FROM tenant_quota_config WHERE tenant_id = ?), 0))`,
		r.ReservedBytes, r.FileCountDelta, r.TenantID, r.ReservedBytes, r.TenantID, DefaultMaxStorageBytes(),
		r.FileCountDelta, r.TenantID, r.FileCountDelta, r.TenantID)
	if err != nil {
		return false, err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return false, err
	}
	return n > 0, nil
}

// insertQuotaUsageRowTx creates the zero-valued usage row for a brand-new
// tenant. Duplicate-key 1062 is handled here, not via isMetaLockConflictError:
// two first-uploads can both see a missing row and race the INSERT; the loser
// treats 1062 as "the row now exists" and re-runs the UPDATE in this same
// transaction. The UPDATE is a current read, so it waits for the winner's
// insert lock and then claims or fails quota.
//
// That in-place 1062 path is not deadlock-free in every cold-path schedule.
// Two concurrent first-uploads can still 1213 on the missing-PK gap lock vs
// insert-intention, and three or more can recreate an S→X cycle after the
// winner commits (two losers both take the duplicate-check S lock, then both
// request X). Those residuals are absorbed by retryMetaLockConflict. They are
// acceptable because production tenants already have a usage row from
// EnsureQuotaUsageRow, so this INSERT is the rare new-tenant first-upload
// path — not the hot-row S→X pattern this change removes.
func (s *Store) insertQuotaUsageRowTx(ctx context.Context, tx *sql.Tx, tenantID string) error {
	_, err := tx.ExecContext(ctx,
		`INSERT INTO tenant_quota_usage (tenant_id) VALUES (?)`, tenantID)
	if err != nil && !isDuplicateEntry(err) {
		return err
	}
	return nil
}

func retryMetaLockConflict(ctx context.Context, tenantID, metricOperation, logOperation string, fn func() error) error {
	var lastErr error
	for attempt := 1; attempt <= metaLockConflictRetryAttempts; attempt++ {
		err := fn()
		if err == nil || !isMetaLockConflictError(err) {
			return err
		}
		lastErr = err
		if attempt == metaLockConflictRetryAttempts {
			break
		}
		logger.Warn(ctx, "meta_lock_conflict_retry",
			zap.String("operation", logOperation),
			zap.Int("attempt", attempt),
			zap.Error(err))
		metrics.RecordTenantOperationCountWithOrg(tenantID, "", "central_quota", metricOperation, "lock_conflict_retry")
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(time.Duration(attempt) * metaLockConflictRetryBase):
		}
	}
	return fmt.Errorf("%w: %w", ErrQuotaReservationBusy, lastErr)
}

func isMetaLockConflictError(err error) bool {
	if err == nil {
		return false
	}
	var mysqlErr *mysql.MySQLError
	if errors.As(err, &mysqlErr) {
		if string(mysqlErr.SQLState[:]) == "40001" {
			return true
		}
		switch mysqlErr.Number {
		case 1213, 1205:
			return true
		case 9007:
			return strings.Contains(strings.ToLower(mysqlErr.Message), "write conflict")
		}
		return false
	}
	msg := err.Error()
	lower := strings.ToLower(msg)
	return (strings.Contains(msg, "Error 1213") && strings.Contains(msg, "Deadlock found")) ||
		(strings.Contains(msg, "ERROR 1213") && strings.Contains(msg, "Deadlock found")) ||
		(strings.Contains(msg, "Error 1205") && strings.Contains(msg, "Lock wait timeout exceeded")) ||
		(strings.Contains(msg, "ERROR 1205") && strings.Contains(msg, "Lock wait timeout exceeded")) ||
		strings.Contains(msg, "SQLSTATE 40001") ||
		((strings.Contains(msg, "Error 9007 (") || strings.Contains(msg, "ERROR 9007 (")) &&
			strings.Contains(lower, "write conflict"))
}

// errQuotaUsageRowMissing is returned by uploadReservationQuotaErrorTx when
// the 0-row UPDATE was caused by a missing tenant_quota_usage row rather
// than a quota limit. Callers bootstrap the row and retry the claim.
var errQuotaUsageRowMissing = errors.New("tenant_quota_usage row missing")

func (s *Store) uploadReservationQuotaErrorTx(ctx context.Context, tx *sql.Tx, r *UploadReservation) error {
	var storageBytes, reservedBytes, fileCount int64
	if err := tx.QueryRowContext(ctx,
		`SELECT storage_bytes, reserved_bytes, file_count
		 FROM tenant_quota_usage WHERE tenant_id = ?`, r.TenantID,
	).Scan(&storageBytes, &reservedBytes, &fileCount); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return errQuotaUsageRowMissing
		}
		return err
	}
	var maxStorageBytes, maxFileCount sql.NullInt64
	if err := tx.QueryRowContext(ctx,
		`SELECT max_storage_bytes, max_file_count
		 FROM tenant_quota_config WHERE tenant_id = ?`, r.TenantID,
	).Scan(&maxStorageBytes, &maxFileCount); err != nil && !errors.Is(err, sql.ErrNoRows) {
		return err
	}
	storageLimit := DefaultMaxStorageBytes()
	if maxStorageBytes.Valid {
		storageLimit = maxStorageBytes.Int64
	}
	if storageLimit > 0 && storageBytes+reservedBytes+r.ReservedBytes > storageLimit {
		return ErrStorageQuotaExceeded
	}
	if r.FileCountDelta > 0 && maxFileCount.Valid && maxFileCount.Int64 > 0 && fileCount+r.FileCountDelta > maxFileCount.Int64 {
		return ErrFileCountQuotaExceeded
	}
	return nil
}

// --- FileMeta operations ---

// UpsertFileMeta creates or updates a file's quota-relevant metadata.
func (s *Store) UpsertFileMeta(ctx context.Context, fm *FileMeta) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "upsert_file_meta", start, &err)

	_, err = s.db.ExecContext(ctx,
		`INSERT INTO tenant_file_meta (tenant_id, file_id, size_bytes, is_media, storage_type, layer_owned)
		 VALUES (?, ?, ?, ?, ?, ?)
		 ON DUPLICATE KEY UPDATE
		   size_bytes = VALUES(size_bytes),
		   is_media = VALUES(is_media),
		   storage_type = VALUES(storage_type),
		   layer_owned = VALUES(layer_owned)`,
		fm.TenantID, fm.FileID, fm.SizeBytes, boolToInt(fm.IsMedia), nullableStorageType(fm.StorageType), boolToInt(fm.LayerOwned))
	return err
}

// UpsertFileMetaTx creates or updates a file's quota metadata inside a transaction.
func (s *Store) UpsertFileMetaTx(tx *sql.Tx, fm *FileMeta) error {
	_, err := tx.Exec(
		`INSERT INTO tenant_file_meta (tenant_id, file_id, size_bytes, is_media, storage_type, layer_owned)
		 VALUES (?, ?, ?, ?, ?, ?)
		 ON DUPLICATE KEY UPDATE
		   size_bytes = VALUES(size_bytes),
		   is_media = VALUES(is_media),
		   storage_type = VALUES(storage_type),
		   layer_owned = VALUES(layer_owned)`,
		fm.TenantID, fm.FileID, fm.SizeBytes, boolToInt(fm.IsMedia), nullableStorageType(fm.StorageType), boolToInt(fm.LayerOwned))
	if err != nil || fm.StorageType != nil {
		return err
	}
	return invalidateStorageTierReadinessTx(tx, fm.TenantID, "unclassified core storage tier")
}

// GetFileMeta returns a file's quota-relevant metadata from the server DB.
func (s *Store) GetFileMeta(ctx context.Context, tenantID, fileID string) (*FileMeta, error) {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "get_file_meta", start, &err)

	return scanFileMeta(s.db.QueryRowContext(ctx,
		`SELECT size_bytes, is_media, storage_type, layer_owned, created_at, updated_at
		 FROM tenant_file_meta WHERE tenant_id = ? AND file_id = ?`,
		tenantID, fileID,
	), tenantID, fileID)
}

// GetFileMetaForUpdateTx returns quota-relevant file metadata and locks the row
// for the caller's central quota transaction.
func (s *Store) GetFileMetaForUpdateTx(tx *sql.Tx, tenantID, fileID string) (*FileMeta, error) {
	return scanFileMeta(tx.QueryRow(
		`SELECT size_bytes, is_media, storage_type, layer_owned, created_at, updated_at
		 FROM tenant_file_meta WHERE tenant_id = ? AND file_id = ? FOR UPDATE`,
		tenantID, fileID,
	), tenantID, fileID)
}

func scanFileMeta(row interface {
	Scan(dest ...any) error
}, tenantID, fileID string) (*FileMeta, error) {
	fm := &FileMeta{TenantID: tenantID, FileID: fileID}
	var isMedia int
	var storageType sql.NullString
	var layerOwned int
	err := row.Scan(&fm.SizeBytes, &isMedia, &storageType, &layerOwned, &fm.CreatedAt, &fm.UpdatedAt)
	if err == sql.ErrNoRows {
		err = ErrNotFound
		return nil, err
	}
	if err != nil {
		return nil, err
	}
	fm.IsMedia = isMedia == 1
	fm.LayerOwned = layerOwned == 1
	if storageType.Valid {
		value := StorageType(storageType.String)
		fm.StorageType = &value
	}
	return fm, nil
}

func nullableStorageType(storageType *StorageType) any {
	if storageType == nil {
		return nil
	}
	return string(*storageType)
}

// DeleteFileMeta removes a file's metadata from the server DB.
func (s *Store) DeleteFileMeta(ctx context.Context, tenantID, fileID string) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "delete_file_meta", start, &err)

	_, err = s.db.ExecContext(ctx,
		`DELETE FROM tenant_file_meta WHERE tenant_id = ? AND file_id = ?`,
		tenantID, fileID)
	return err
}

// DeleteFileMetaTx removes a file's metadata inside a transaction.
func (s *Store) DeleteFileMetaTx(tx *sql.Tx, tenantID, fileID string) error {
	_, err := tx.Exec(
		`DELETE FROM tenant_file_meta WHERE tenant_id = ? AND file_id = ?`,
		tenantID, fileID)
	return err
}

// DeleteFileMetaIfExistsTx removes a file's metadata inside a transaction and
// reports whether a row existed. Callers use the boolean to make delete-side
// quota counter updates idempotent across task retries.
func (s *Store) DeleteFileMetaIfExistsTx(tx *sql.Tx, tenantID, fileID string) (bool, error) {
	res, err := tx.Exec(
		`DELETE FROM tenant_file_meta WHERE tenant_id = ? AND file_id = ?`,
		tenantID, fileID)
	if err != nil {
		return false, err
	}
	rowsAffected, err := res.RowsAffected()
	if err != nil {
		return false, err
	}
	return rowsAffected > 0, nil
}

// InsertLayerStorageMetaIfAbsentTx records a physical fs-layer payload once.
// The immutable entry identity makes both mutation replay and lazy scan
// idempotent without updating an existing row.
func (s *Store) InsertLayerStorageMetaIfAbsentTx(tx *sql.Tx, storage *LayerStorageMeta) (bool, error) {
	var storageType *StorageType
	if storage.StorageType == StorageTypeDB9 || storage.StorageType == StorageTypeS3 {
		value := storage.StorageType
		storageType = &value
	}
	res, err := tx.Exec(`INSERT IGNORE INTO tenant_layer_storage_meta
		(tenant_id, layer_id, path_hash, entry_seq, size_bytes, storage_type)
		VALUES (?, ?, ?, ?, ?, ?)`, storage.TenantID, storage.LayerID, storage.PathHash,
		storage.EntrySeq, storage.SizeBytes, nullableStorageType(storageType))
	if err != nil {
		return false, err
	}
	n, err := res.RowsAffected()
	if err != nil || n == 0 || storageType != nil {
		return n > 0, err
	}
	if err := invalidateStorageTierReadinessTx(tx, storage.TenantID, "unclassified fs-layer storage tier"); err != nil {
		return false, err
	}
	return true, nil
}

func invalidateStorageTierReadinessTx(tx *sql.Tx, tenantID, reason string) error {
	_, err := tx.Exec(`UPDATE tenant_storage_tier_scan SET
		status = 'blocked', full_scan_completed_at = NULL, lease_owner = '', lease_until = NULL, last_error = ?
		WHERE tenant_id = ? AND full_scan_completed_at IS NOT NULL`, reason, tenantID)
	return err
}

// --- Upload reservation operations ---

// InsertUploadReservation creates a new active upload reservation.
func (s *Store) InsertUploadReservation(ctx context.Context, r *UploadReservation) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "insert_upload_reservation", start, &err)

	_, err = s.db.ExecContext(ctx,
		`INSERT INTO tenant_upload_reservations (tenant_id, upload_id, reserved_bytes, file_count_delta, target_path, status, expires_at)
		 VALUES (?, ?, ?, ?, ?, 'active', ?)`,
		r.TenantID, r.UploadID, r.ReservedBytes, r.FileCountDelta, r.TargetPath, r.ExpiresAt)
	return err
}

// UpdateUploadReservationStatus updates a reservation's status.
func (s *Store) UpdateUploadReservationStatus(ctx context.Context, tenantID, uploadID, status string) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "update_upload_reservation_status", start, &err)

	_, err = s.db.ExecContext(ctx,
		`UPDATE tenant_upload_reservations SET status = ? WHERE tenant_id = ? AND upload_id = ? AND status IN ('active', 'completing')`,
		status, tenantID, uploadID)
	return err
}

// UpdateUploadReservationStatusTx updates a reservation's status inside a transaction.
func (s *Store) UpdateUploadReservationStatusTx(ctx context.Context, tx *sql.Tx, tenantID, uploadID, status string) error {
	_, err := tx.ExecContext(ctx,
		`UPDATE tenant_upload_reservations SET status = ? WHERE tenant_id = ? AND upload_id = ? AND status IN ('active', 'completing')`,
		status, tenantID, uploadID)
	return err
}

// AbortActiveReservationTx atomically claims an active/completing reservation
// for abort and reports the counters that must be released by the caller in the
// same transaction.
func (s *Store) AbortActiveReservationTx(ctx context.Context, tx *sql.Tx, tenantID, uploadID string) (aborted bool, reservedBytes, fileCountDelta int64, err error) {
	var bytes, files sql.NullInt64
	if err := tx.QueryRowContext(ctx,
		`SELECT reserved_bytes, file_count_delta FROM tenant_upload_reservations
		 WHERE tenant_id = ? AND upload_id = ? AND status IN ('active', 'completing') FOR UPDATE`,
		tenantID, uploadID).Scan(&bytes, &files); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return false, 0, 0, nil
		}
		return false, 0, 0, err
	}
	res, err := tx.ExecContext(ctx,
		`UPDATE tenant_upload_reservations SET status = 'aborted'
		 WHERE tenant_id = ? AND upload_id = ? AND status IN ('active', 'completing')`,
		tenantID, uploadID)
	if err != nil {
		return false, 0, 0, err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return false, 0, 0, nil
	}
	return true, bytes.Int64, files.Int64, nil
}

// SettleActiveReservationTx transitions a reservation from 'active' to the
// given terminal status inside a transaction, reporting whether a row was
// actually updated. Callers use this to decide — inside the same tx — whether
// to apply reserved_bytes-related counter changes.
//
// settled=true means an 'active' row was atomically flipped to status; the
// caller should do the reserved→storage transfer. settled=false signals "do
// not touch reserved_bytes" and covers three indistinguishable sub-cases:
//   - fail-open initiate: no reservation row was ever written because the
//     server DB was down at initiate time.
//   - already-settled terminal row: a concurrent replay or abort flipped
//     the row before this tx ran.
//   - apply-time expiry sweep race: the expiry sweep released reserved_bytes
//     and marked the row aborted between initiate and apply.
//
// This is the atomic replacement for a separate GetUploadReservation lookup
// outside the transaction, which has two failure modes we care about:
//   - Transient error: we cannot tell whether a reservation exists, so we
//     would otherwise have to bail out and silently drop the paired mutation.
//   - TOCTOU: between lookup and apply another worker could settle the row.
func (s *Store) SettleActiveReservationTx(ctx context.Context, tx *sql.Tx, tenantID, uploadID, status string) (settled bool, fileCountDelta int64, err error) {
	var delta sql.NullInt64
	if err := tx.QueryRowContext(ctx,
		`SELECT file_count_delta FROM tenant_upload_reservations
		 WHERE tenant_id = ? AND upload_id = ? AND status IN ('active', 'completing') FOR UPDATE`,
		tenantID, uploadID).Scan(&delta); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return false, 0, nil
		}
		return false, 0, err
	}
	res, err := tx.ExecContext(ctx,
		`UPDATE tenant_upload_reservations SET status = ? WHERE tenant_id = ? AND upload_id = ? AND status IN ('active', 'completing')`,
		status, tenantID, uploadID)
	if err != nil {
		return false, 0, err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return false, 0, nil
	}
	return true, delta.Int64, nil
}

// GetUploadReservation returns a single reservation.
func (s *Store) GetUploadReservation(ctx context.Context, tenantID, uploadID string) (*UploadReservation, error) {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "get_upload_reservation", start, &err)

	r := &UploadReservation{TenantID: tenantID, UploadID: uploadID}
	err = s.db.QueryRowContext(ctx,
		`SELECT reserved_bytes, file_count_delta, target_path, status, expires_at, created_at, updated_at
		 FROM tenant_upload_reservations WHERE tenant_id = ? AND upload_id = ?`,
		tenantID, uploadID,
	).Scan(&r.ReservedBytes, &r.FileCountDelta, &r.TargetPath, &r.Status, &r.ExpiresAt, &r.CreatedAt, &r.UpdatedAt)
	if err == sql.ErrNoRows {
		err = ErrNotFound
		return nil, err
	}
	if err != nil {
		return nil, err
	}
	return r, nil
}

func (s *Store) HasActiveUploadReservations(ctx context.Context, tenantID string) (bool, error) {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "has_active_upload_reservations", start, &err)

	var one int
	err = s.db.QueryRowContext(ctx,
		`SELECT 1
		 FROM tenant_upload_reservations
		 WHERE tenant_id = ? AND status IN ('active', 'completing') AND expires_at > ?
		 LIMIT 1`, tenantID, time.Now().UTC()).Scan(&one)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	return true, nil
}

func (s *Store) AbortActiveUploadReservations(ctx context.Context, tenantID string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "abort_active_upload_reservations", start, &err)

	err = s.InTx(ctx, func(tx *sql.Tx) error {
		var totalBytes, totalFiles sql.NullInt64
		if err := tx.QueryRowContext(ctx,
			`SELECT COALESCE(SUM(reserved_bytes), 0), COALESCE(SUM(file_count_delta), 0)
			 FROM tenant_upload_reservations
			 WHERE tenant_id = ? AND status IN ('active', 'completing')`, tenantID).Scan(&totalBytes, &totalFiles); err != nil {
			return err
		}
		if totalBytes.Int64 != 0 || totalFiles.Int64 != 0 {
			res, err := tx.ExecContext(ctx,
				`UPDATE tenant_quota_usage
				 SET reserved_bytes = GREATEST(reserved_bytes - ?, 0),
				     file_count = GREATEST(file_count - ?, 0)
				 WHERE tenant_id = ?`, totalBytes.Int64, totalFiles.Int64, tenantID)
			if err != nil {
				return err
			}
			if n, _ := res.RowsAffected(); n == 0 {
				return fmt.Errorf("tenant_quota_usage row missing for tenant %s", tenantID)
			}
		}
		_, err := tx.ExecContext(ctx,
			`UPDATE tenant_upload_reservations SET status = 'aborted'
			 WHERE tenant_id = ? AND status IN ('active', 'completing')`, tenantID)
		return err
	})
	return err
}

// ExpireActiveReservations marks expired active/completing reservations as
// aborted and returns the total bytes released. This is called by the expiry
// sweep worker. All changes are applied in a single transaction to prevent
// double-release if the process crashes mid-sweep. Completing reservations are
// included so a crash after mark-completing but before finalize settle does not
// leak reserved_bytes forever.
func (s *Store) ExpireActiveReservations(ctx context.Context) (int64, error) {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "expire_active_reservations", start, &err)

	now := time.Now().UTC()
	var totalReleased int64

	err = s.InTx(ctx, func(tx *sql.Tx) error {
		// Collect tenant-level totals so we can adjust quota counters.
		rows, err := tx.QueryContext(ctx,
			`SELECT tenant_id, SUM(reserved_bytes), SUM(file_count_delta)
			 FROM tenant_upload_reservations
			 WHERE status IN ('active', 'completing') AND expires_at < ?
			 GROUP BY tenant_id`, now)
		if err != nil {
			return err
		}
		defer func() { _ = rows.Close() }()

		type tenantRelease struct {
			tenantID      string
			releasedBytes int64
			releasedFiles int64
		}
		var releases []tenantRelease
		for rows.Next() {
			var tr tenantRelease
			if err := rows.Scan(&tr.tenantID, &tr.releasedBytes, &tr.releasedFiles); err != nil {
				return err
			}
			releases = append(releases, tr)
		}
		if err := rows.Err(); err != nil {
			return err
		}

		for _, tr := range releases {
			res, err := tx.Exec(
				`UPDATE tenant_quota_usage
				 SET reserved_bytes = reserved_bytes + ?,
				     file_count = GREATEST(file_count - ?, 0)
				 WHERE tenant_id = ?`,
				-tr.releasedBytes, tr.releasedFiles, tr.tenantID)
			if err != nil {
				return fmt.Errorf("release reserved bytes for %s: %w", tr.tenantID, err)
			}
			if n, _ := res.RowsAffected(); n == 0 {
				return fmt.Errorf("tenant_quota_usage row missing for tenant %s", tr.tenantID)
			}
			totalReleased += tr.releasedBytes
		}

		_, err = tx.ExecContext(ctx,
			`UPDATE tenant_upload_reservations SET status = 'aborted'
			 WHERE status IN ('active', 'completing') AND expires_at < ?`, now)
		return err
	})
	return totalReleased, err
}

// --- LLM usage operations ---

// InsertCentralLLMUsage records a billable LLM call in the central ledger.
func (s *Store) InsertCentralLLMUsage(ctx context.Context, r *LLMUsageRecord) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "insert_central_llm_usage", start, &err)

	_, err = s.db.ExecContext(ctx,
		`INSERT INTO tenant_llm_usage (tenant_id, task_type, task_id, cost_millicents, raw_units, raw_unit_type)
		 VALUES (?, ?, ?, ?, ?, ?)`,
		r.TenantID, r.TaskType, r.TaskID, r.CostMillicents, r.RawUnits, r.RawUnitType)
	return err
}

// InsertCentralLLMUsageTx records a billable LLM call in the central ledger inside a transaction.
func (s *Store) InsertCentralLLMUsageTx(tx *sql.Tx, r *LLMUsageRecord) error {
	_, err := tx.Exec(
		`INSERT INTO tenant_llm_usage (tenant_id, task_type, task_id, cost_millicents, raw_units, raw_unit_type)
		 VALUES (?, ?, ?, ?, ?, ?)`,
		r.TenantID, r.TaskType, r.TaskID, r.CostMillicents, r.RawUnits, r.RawUnitType)
	return err
}

// IncrMonthlyLLMCost atomically increments the pre-aggregated monthly cost counter.
func (s *Store) IncrMonthlyLLMCost(ctx context.Context, tenantID string, costMC int64) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "incr_monthly_llm_cost", start, &err)

	now := time.Now().UTC()
	monthStart := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, time.UTC)
	_, err = s.db.ExecContext(ctx,
		`INSERT INTO tenant_monthly_llm_cost (tenant_id, month_start, total_mc)
		 VALUES (?, ?, ?)
		 ON DUPLICATE KEY UPDATE total_mc = total_mc + VALUES(total_mc)`,
		tenantID, monthStart, costMC)
	return err
}

// IncrMonthlyLLMCostTx atomically increments the pre-aggregated monthly cost counter inside a transaction.
func (s *Store) IncrMonthlyLLMCostTx(tx *sql.Tx, tenantID string, costMC int64) error {
	now := time.Now().UTC()
	monthStart := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, time.UTC)
	_, err := tx.Exec(
		`INSERT INTO tenant_monthly_llm_cost (tenant_id, month_start, total_mc)
		 VALUES (?, ?, ?)
		 ON DUPLICATE KEY UPDATE total_mc = total_mc + VALUES(total_mc)`,
		tenantID, monthStart, costMC)
	return err
}

// --- Mutation log operations ---

// InsertMutationLog writes a pending mutation to the outbox.
func (s *Store) InsertMutationLog(ctx context.Context, entry *MutationLogEntry) (int64, error) {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "insert_mutation_log", start, &err)

	res, err := s.db.ExecContext(ctx,
		`INSERT INTO quota_mutation_log (tenant_id, mutation_type, mutation_data, status)
		 VALUES (?, ?, ?, 'pending')`,
		entry.TenantID, entry.MutationType, entry.MutationData)
	if err != nil {
		return 0, err
	}
	return res.LastInsertId()
}

// listPendingMutationsOrgJoins is the shared org-resolution tail for the
// pending-mutation queries: org id comes from db_pool for shared-provider
// tenants and from tenant_tidbcloud_org_bindings otherwise.
const listPendingMutationsOrgJoins = `
		 LEFT JOIN tenants t ON t.id = q.tenant_id
		 LEFT JOIN tenant_tidbcloud_org_bindings b ON b.tenant_id = q.tenant_id
		 LEFT JOIN fs_registry f ON f.tenant_id = q.tenant_id
		 LEFT JOIN tenant_placements p ON p.fs_id = f.fs_id
		 LEFT JOIN db_pool d ON d.db_id = p.db_id`

const listPendingMutationsOrgSelect = `
		        COALESCE(CASE WHEN t.provider = ? THEN d.org_id ELSE b.organization_id END, '')`

// listPendingMutationsForTenantSQL pages one tenant's pending mutations by id.
// FORCE INDEX pins the pending-only index — see listPendingMutationsForTenant.
const listPendingMutationsForTenantSQL = `SELECT q.id, q.tenant_id, ` + listPendingMutationsOrgSelect + `,
		        q.mutation_type, q.mutation_data, q.status, q.retry_count, q.created_at, q.applied_at
		 FROM quota_mutation_log q FORCE INDEX (idx_pending_tenant_age)` + listPendingMutationsOrgJoins + `
		 WHERE q.status = 'pending' AND q.tenant_id = ? AND q.created_at < ? AND q.id > ?
		 ORDER BY q.id ASC
		 LIMIT ?`

// pendingMutationTenantLimit caps how many distinct backlogged tenants one
// ListPendingMutations call spans. Must stay <= the replay worker's
// replayBatchLimit (100), otherwise the floor share drops to 1 and later
// tenants can again be crowded out of a batch. The per-tenant share of the
// batch limit is derived from the actual tenant count, so this only bounds
// query fan-out.
const pendingMutationTenantLimit = 32

// ListPendingMutations returns pending mutations older than the given age.
// Entries for one tenant are id-ascending (the per-tenant
// sequential-processing contract the replay barrier relies on); tenants are
// not necessarily contiguous.
//
// Tenant selection is fair: tenants are picked oldest-backlog-first and get
// even shares of the batch limit, with top-up passes redistributing unused
// shares so early tenants with deep backlogs cannot exhaust the limit before
// later tenants are queried. The previous implementation ordered a single
// flat query by (tenant_id, id) with LIMIT, which let one high-volume tenant
// whose id sorts early fill every batch and starve alphabetically-later
// tenants indefinitely (observed in prod: a tenant with thousands of pendings
// blocked all letter-prefixed tenants for >10 minutes, tripping the
// pending-age alert while the replay worker kept busy).
func (s *Store) ListPendingMutations(ctx context.Context, minAge time.Duration, limit int) ([]MutationLogEntry, error) {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "list_pending_mutations", start, &err)

	if limit <= 0 {
		limit = 100
	}
	cutoff := time.Now().UTC().Add(-minAge)

	tenantRows, err := s.db.QueryContext(ctx,
		`SELECT tenant_id
		 FROM quota_mutation_log FORCE INDEX (idx_pending_tenant_age)
		 WHERE status = 'pending' AND created_at < ?
		 GROUP BY tenant_id
		 ORDER BY MIN(created_at), tenant_id
		 LIMIT ?`, cutoff, pendingMutationTenantLimit)
	if err != nil {
		return nil, err
	}
	var tenantIDs []string
	for tenantRows.Next() {
		var tenantID string
		if err = tenantRows.Scan(&tenantID); err != nil {
			_ = tenantRows.Close()
			return nil, err
		}
		tenantIDs = append(tenantIDs, tenantID)
	}
	if err = tenantRows.Err(); err != nil {
		_ = tenantRows.Close()
		return nil, err
	}
	if err = tenantRows.Close(); err != nil {
		return nil, err
	}
	if len(tenantIDs) == 0 {
		return nil, nil
	}

	// Floor (not ceil) share: the first pass must leave room for every
	// selected tenant — a ceil share (e.g. 4 rows each across 32 tenants with
	// limit 100) lets the first 25 deep-backlog tenants exhaust the batch
	// before tenants 26-32 are even queried, recreating the starvation this
	// lister exists to fix. Top-up passes then redistribute the remainder
	// oldest-first, and tenants that return fewer rows than requested are
	// dropped from the rotation so later passes do not re-query them.
	perTenant := limit / len(tenantIDs)
	if perTenant < 1 {
		perTenant = 1
	}
	lastID := make(map[string]int64, len(tenantIDs))
	active := tenantIDs
	var entries []MutationLogEntry
	for len(entries) < limit && len(active) > 0 {
		progress := false
		var stillBacklogged []string
		for _, tenantID := range active {
			remaining := limit - len(entries)
			if remaining <= 0 {
				break
			}
			batchLimit := perTenant
			if batchLimit > remaining {
				batchLimit = remaining
			}
			var batch []MutationLogEntry
			batch, err = s.listPendingMutationsForTenant(ctx, tenantID, lastID[tenantID], cutoff, batchLimit)
			if err != nil {
				return nil, err
			}
			if len(batch) == 0 {
				continue
			}
			progress = true
			lastID[tenantID] = batch[len(batch)-1].ID
			entries = append(entries, batch...)
			if len(batch) == batchLimit {
				// The tenant may hold more rows beyond this share.
				stillBacklogged = append(stillBacklogged, tenantID)
			}
		}
		active = stillBacklogged
		if !progress {
			break
		}
	}
	return entries, nil
}

// listPendingMutationsForTenant fetches up to limit pending mutations for one
// tenant with id > afterID, ordered by id. Callers page through a tenant's
// backlog by feeding the last returned id back as afterID.
//
// FORCE INDEX (idx_pending_tenant_age) pins access to the pending-only index:
// without it the optimizer can pick idx_tenant_order (tenant_id, id), which
// matches the id range + ORDER BY shape perfectly but walks the tenant's
// entire applied history (150M+ rows cluster-wide) filtering status.
func (s *Store) listPendingMutationsForTenant(ctx context.Context, tenantID string, afterID int64, cutoff time.Time, limit int) ([]MutationLogEntry, error) {
	rows, err := s.db.QueryContext(ctx,
		listPendingMutationsForTenantSQL, tidbCloudNativeSharedProvider, tenantID, cutoff, afterID, limit)
	if err != nil {
		return nil, err
	}
	var entries []MutationLogEntry
	for rows.Next() {
		var e MutationLogEntry
		if err := rows.Scan(&e.ID, &e.TenantID, &e.TiDBCloudOrgID, &e.MutationType, &e.MutationData, &e.Status, &e.RetryCount, &e.CreatedAt, &e.AppliedAt); err != nil {
			_ = rows.Close()
			return nil, err
		}
		entries = append(entries, e)
	}
	if err := rows.Err(); err != nil {
		_ = rows.Close()
		return nil, err
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	return entries, nil
}

// PurgeAppliedMutations deletes up to limit applied mutation log rows that
// were both created and applied before cutoff, returning the number of
// deleted rows. Pending and failed rows are never touched (failed rows are
// kept for diagnosis). The applied_at predicate keeps rows that were created
// long ago but only recently converged (e.g. after a backlog): from a
// debugging/audit perspective they are still fresh.
//
// The table carries no index on (status, applied_at), so a naive
// DELETE ... WHERE status='applied' AND applied_at < ? would scan the whole
// applied prefix of idx_pending (150M+ rows in prod) per batch. Instead the
// row set is bounded by a primary-key boundary: created_at correlates with
// the auto-increment id, so binary searching the PK for the created_at
// cutoff (O(log n) point lookups) yields an id range the DELETE can consume
// as a PK range scan, filtering status/created_at/applied_at within it. A
// slightly-off boundary only changes how many rows a batch covers, never
// which rows are eligible. (Adding a purge index was considered and declined
// for this change: an online ADD INDEX on a 64GB table is a heavier rollout
// than this PR's no-schema-change scope; revisit if purge I/O shows up.)
func (s *Store) PurgeAppliedMutations(ctx context.Context, cutoff time.Time, limit int) (int64, error) {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "purge_applied_mutations", start, &err)

	if limit <= 0 {
		limit = 5000
	}
	boundary, err := s.mutationLogPurgeBoundary(ctx, cutoff)
	if err != nil {
		return 0, err
	}
	if boundary == 0 {
		return 0, nil
	}
	res, err := s.db.ExecContext(ctx,
		`DELETE FROM quota_mutation_log WHERE id < ? AND status = 'applied' AND created_at < ? AND applied_at < ? LIMIT ?`,
		boundary, cutoff, cutoff, limit)
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}

// mutationLogPurgeBoundary returns the smallest id whose row is at or after
// cutoff — i.e. every applied row with id < boundary is a purge candidate.
// Returns 0 when the table is empty (aggregates over an empty table yield a
// single NULL row, hence the COALESCE). Gaps below the boundary (already
// purged ids) are treated as before-cutoff: rows are only ever deleted below
// the boundary, so a missing id can never belong to the retained side.
// The search range is [minID, maxID+1) so a fully-expired table converges to
// maxID+1 and the DELETE still covers the max-id row.
func (s *Store) mutationLogPurgeBoundary(ctx context.Context, cutoff time.Time) (int64, error) {
	var minID, maxID int64
	err := s.db.QueryRowContext(ctx,
		`SELECT COALESCE(MIN(id), 0), COALESCE(MAX(id), 0) FROM quota_mutation_log`).Scan(&minID, &maxID)
	if err != nil {
		return 0, err
	}
	if maxID == 0 {
		return 0, nil
	}
	lo, hi := minID, maxID+1
	for lo < hi {
		mid := lo + (hi-lo)/2
		var createdAt time.Time
		err := s.db.QueryRowContext(ctx, `SELECT created_at FROM quota_mutation_log WHERE id = ?`, mid).Scan(&createdAt)
		if errors.Is(err, sql.ErrNoRows) {
			lo = mid + 1
			continue
		}
		if err != nil {
			return 0, err
		}
		if createdAt.Before(cutoff) {
			lo = mid + 1
		} else {
			hi = mid
		}
	}
	return lo, nil
}

// Aggregate the mutation log before resolving tenant organization metadata so
// each pending tenant, rather than each pending mutation, performs the joins.
const observePendingMutationsSQL = `WITH pending_mutations AS (
	SELECT tenant_id, COUNT(*) AS pending_count, MIN(created_at) AS oldest_created_at
		FROM quota_mutation_log FORCE INDEX (idx_pending_tenant_age)
		WHERE status = 'pending'
		GROUP BY tenant_id
)
SELECT pending.tenant_id,
	COALESCE(CASE WHEN t.provider = ? THEN d.org_id ELSE b.organization_id END, ''),
	pending.pending_count, pending.oldest_created_at
	FROM pending_mutations pending
	LEFT JOIN tenants t ON t.id = pending.tenant_id
	LEFT JOIN tenant_tidbcloud_org_bindings b ON b.tenant_id = pending.tenant_id
	LEFT JOIN fs_registry f ON f.tenant_id = pending.tenant_id
	LEFT JOIN tenant_placements p ON p.fs_id = f.fs_id
	LEFT JOIN db_pool d ON d.db_id = p.db_id
	ORDER BY pending.tenant_id`

// ObservePendingMutations returns per-tenant pending mutation backlog and age.
func (s *Store) ObservePendingMutations(ctx context.Context) ([]MutationBacklogObservation, error) {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "observe_pending_mutations", start, &err)

	rows, err := s.db.QueryContext(ctx, observePendingMutationsSQL, tidbCloudNativeSharedProvider)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	now := time.Now().UTC()
	var out []MutationBacklogObservation
	for rows.Next() {
		var obs MutationBacklogObservation
		var oldest time.Time
		if err = rows.Scan(&obs.TenantID, &obs.TiDBCloudOrgID, &obs.PendingCount, &oldest); err != nil {
			return out, err
		}
		age := now.Sub(oldest.UTC()).Seconds()
		if age < 0 {
			age = 0
		}
		obs.OldestPendingAgeSeconds = age
		out = append(out, obs)
	}
	err = rows.Err()
	return out, err
}

// HasPendingFileMutation reports whether an unapplied create/overwrite
// mutation still exists for fileID. GC uses this to avoid acking central cleanup
// before fail-open quota mutations have converged.
func (s *Store) HasPendingFileMutation(ctx context.Context, tenantID, fileID string) (bool, error) {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "has_pending_file_mutation", start, &err)

	var one int
	err = s.db.QueryRowContext(ctx,
		`SELECT 1
		 FROM quota_mutation_log
		 WHERE tenant_id = ?
		   AND status = 'pending'
		   AND mutation_type IN ('file_create', 'file_overwrite')
		   AND JSON_UNQUOTE(JSON_EXTRACT(mutation_data, '$.file_id')) = ?
		 LIMIT 1`, tenantID, fileID).Scan(&one)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	return true, nil
}

// MarkMutationApplied marks a mutation log entry as applied within a transaction.
// The WHERE clause includes status = 'pending' so that concurrent replay workers
// cannot both apply the same mutation: the second worker's UPDATE will affect 0 rows,
// causing the transaction to roll back and preventing double-apply of counter mutations.
func (s *Store) MarkMutationAppliedTx(tx *sql.Tx, id int64) error {
	res, err := tx.Exec(
		`UPDATE quota_mutation_log SET status = 'applied', applied_at = ? WHERE id = ? AND status = 'pending'`,
		time.Now().UTC(), id)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		var status string
		if err := tx.QueryRow(`SELECT status FROM quota_mutation_log WHERE id = ?`, id).Scan(&status); err != nil {
			return fmt.Errorf("mutation %d not pending and status lookup failed: %w", id, err)
		}
		if status == "applied" {
			return fmt.Errorf("mutation %d: %w", id, errMutationAlreadyApplied)
		}
		return fmt.Errorf("mutation %d unexpectedly left pending state before mark applied: status=%q", id, status)
	}
	return requeueBlockedStorageTierScanTx(tx, []int64{id})
}

// MarkMutationsAppliedTx marks a batch of mutation log entries as applied
// within a transaction. It mirrors MarkMutationAppliedTx: the WHERE clause
// includes status = 'pending' and the update must affect exactly len(ids)
// rows, so a concurrent replay worker flipping any row to applied first makes
// the whole batch transaction roll back instead of double-applying counter
// mutations. ids is bounded by the dispatcher batch size (<= 128).
func (s *Store) MarkMutationsAppliedTx(tx *sql.Tx, ids []int64) error {
	if len(ids) == 0 {
		return nil
	}
	placeholders := make([]string, len(ids))
	args := make([]any, 0, len(ids)+1)
	args = append(args, time.Now().UTC())
	for i, id := range ids {
		placeholders[i] = "?"
		args = append(args, id)
	}
	res, err := tx.Exec(
		`UPDATE quota_mutation_log SET status = 'applied', applied_at = ? WHERE id IN (`+strings.Join(placeholders, ",")+`) AND status = 'pending'`,
		args...)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if int(n) != len(ids) {
		return fmt.Errorf("batch mark applied affected %d of %d mutations: %w", n, len(ids), errMutationAlreadyApplied)
	}
	return requeueBlockedStorageTierScanTx(tx, ids)
}

func requeueBlockedStorageTierScanTx(tx *sql.Tx, mutationIDs []int64) error {
	if len(mutationIDs) == 0 {
		return nil
	}
	placeholders := make([]string, len(mutationIDs))
	args := make([]any, 0, len(mutationIDs))
	for i, id := range mutationIDs {
		placeholders[i] = "?"
		args = append(args, id)
	}
	// Filter out normal pending/running/completed scans before attempting an
	// UPDATE. Mutation apply already holds file/quota rows, while the scanner
	// locks the scan row first; touching an active scan row here would invert
	// that lock order. A blocked job cannot be claimed by the scanner.
	rows, err := tx.Query(`SELECT DISTINCT q.tenant_id FROM quota_mutation_log q
		JOIN tenant_storage_tier_scan s ON s.tenant_id = q.tenant_id AND s.status = 'blocked'
		WHERE q.id IN (`+strings.Join(placeholders, ",")+`) AND q.status = 'applied'
		  AND q.mutation_type IN ('file_create', 'file_overwrite', 'file_delete', 'upload_complete', 'fs_layer_store')
		ORDER BY q.tenant_id`, args...)
	if err != nil {
		return err
	}
	var tenantIDs []string
	for rows.Next() {
		var tenantID string
		if err := rows.Scan(&tenantID); err != nil {
			_ = rows.Close()
			return err
		}
		tenantIDs = append(tenantIDs, tenantID)
	}
	if err := rows.Close(); err != nil {
		return err
	}
	if err := rows.Err(); err != nil {
		return err
	}

	for _, tenantID := range tenantIDs {
		res, err := tx.Exec(`UPDATE tenant_storage_tier_scan s
			SET s.status = 'pending', s.scan_phase = 'core', s.scan_cursor = '',
			    s.layer_path_hash = '', s.layer_entry_seq = 0,
			    s.watermark = (SELECT COALESCE(MAX(q.id), 0) FROM quota_mutation_log q
			                   WHERE q.tenant_id = s.tenant_id
			                     AND q.mutation_type IN ('file_create', 'file_overwrite', 'file_delete', 'upload_complete', 'fs_layer_store')),
			    s.lease_owner = '', s.lease_until = NULL, s.last_error = 'repair mutation applied; rescan required',
			    s.full_scan_completed_at = NULL
			WHERE s.tenant_id = ? AND s.status = 'blocked'
			  AND NOT EXISTS (
				SELECT 1 FROM quota_mutation_log q
				WHERE q.tenant_id = s.tenant_id AND q.status IN ('pending', 'failed')
				  AND q.mutation_type IN ('file_create', 'file_overwrite', 'file_delete', 'upload_complete', 'fs_layer_store')
			  )`, tenantID)
		if err != nil {
			return err
		}
		updated, err := res.RowsAffected()
		if err != nil {
			return err
		}
		if updated == 0 {
			continue
		}
		if _, err := tx.Exec(`INSERT INTO tenant_notify_outbox (tenant_id, work_mask) VALUES (?, ?)`, tenantID, TenantNotifyWorkStorageTierScan); err != nil {
			return err
		}
	}
	return nil
}

// IncrMutationRetry increments the retry count for a pending mutation.
// The WHERE status='pending' guard mirrors MarkMutationAppliedTx and prevents
// any caller (test fakes, backfill CLI, future refactors) from silently flipping
// applied/failed terminal rows back to pending or bumping retry_count on a row
// that has already been settled. If max retries is reached, transitions to
// the terminal 'failed' status. Calls against rows not in 'pending' state
// are no-ops.
func (s *Store) IncrMutationRetry(ctx context.Context, id int64, maxRetries int) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "incr_mutation_retry", start, &err)

	err = s.InTx(ctx, func(tx *sql.Tx) error {
		if _, err := tx.Exec(
			`UPDATE quota_mutation_log
			 SET retry_count = retry_count + 1,
			     status = CASE WHEN retry_count + 1 >= ? THEN 'failed' ELSE 'pending' END
			 WHERE id = ? AND status = 'pending'`, maxRetries, id); err != nil {
			return err
		}
		_, err := tx.Exec(`UPDATE tenant_storage_tier_scan s
			JOIN quota_mutation_log q ON q.tenant_id = s.tenant_id
			SET s.status = 'blocked', s.full_scan_completed_at = NULL,
			    s.lease_owner = '', s.lease_until = NULL, s.last_error = 'storage tier mutation failed'
			WHERE q.id = ? AND q.status = 'failed'
			  AND q.mutation_type IN ('file_create', 'file_overwrite', 'file_delete', 'upload_complete', 'fs_layer_store')`, id)
		return err
	})
	return err
}

// SetQuotaCounters atomically sets the absolute quota counter values for a
// tenant. Used by the backfill CLI to bootstrap counters from tenant DBs.
func (s *Store) SetQuotaCounters(ctx context.Context, tenantID string, storageBytes, mediaFileCount, fileCount int64) error {
	start := time.Now()
	var err error
	defer observeMeta(ctx, "set_quota_counters", start, &err)

	_, err = s.db.ExecContext(ctx,
		`INSERT INTO tenant_quota_usage (tenant_id, storage_bytes, reserved_bytes, file_count, media_file_count)
		 VALUES (?, ?, 0, ?, ?)
		 ON DUPLICATE KEY UPDATE
		   storage_bytes = VALUES(storage_bytes),
		   file_count = VALUES(file_count),
		   media_file_count = VALUES(media_file_count)`,
		tenantID, storageBytes, fileCount, mediaFileCount)
	return err
}

// InTx runs a function within a server DB transaction.
func (s *Store) InTx(ctx context.Context, fn func(tx *sql.Tx) error) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	if err := fn(tx); err != nil {
		_ = tx.Rollback()
		return err
	}
	return tx.Commit()
}