package meta

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

type StorageTierScanJob struct {
	TenantID            string
	Status              string
	Phase               string
	Cursor              string
	LayerPathHash       string
	LayerEntrySeq       int64
	Watermark           int64
	LeaseOwner          string
	LeaseUntil          *time.Time
	AttemptCount        int
	LastError           string
	FullScanCompletedAt *time.Time
	UpdatedAt           time.Time
}

const (
	StorageTierScanPhaseCore  = "core"
	StorageTierScanPhaseLayer = "layer"

	StorageTierScanStatusPending   = "pending"
	StorageTierScanStatusRunning   = "running"
	StorageTierScanStatusBlocked   = "blocked"
	StorageTierScanStatusCompleted = "completed"
)

type StorageTierMutationState struct {
	Pending int64
	Failed  int64
}

type StorageTierScanFile struct {
	FileID      string
	SizeBytes   int64
	IsMedia     bool
	StorageType StorageType
	LayerOwned  bool
}

type StorageTierLayerCursor struct {
	LayerID  string
	PathHash string
	EntrySeq int64
}

type StorageTierScanLayer struct {
	StorageTierLayerCursor
	SizeBytes   int64
	StorageType StorageType
}

func (s *Store) EnsureStorageTierScanJob(ctx context.Context, tenantID string) (bool, error) {
	res, err := s.db.ExecContext(ctx, `INSERT IGNORE INTO tenant_storage_tier_scan
		(tenant_id, watermark)
		SELECT ?, COALESCE(MAX(id), 0) FROM quota_mutation_log WHERE tenant_id = ?`, tenantID, tenantID)
	if err != nil {
		return false, fmt.Errorf("ensure storage tier scan job: %w", err)
	}
	n, err := res.RowsAffected()
	return n > 0, err
}

func (s *Store) GetStorageTierScanJob(ctx context.Context, tenantID string) (*StorageTierScanJob, error) {
	job := &StorageTierScanJob{TenantID: tenantID}
	var leaseUntil, completedAt sql.NullTime
	var lastError sql.NullString
	err := s.db.QueryRowContext(ctx, `SELECT status, scan_phase, scan_cursor, layer_path_hash, layer_entry_seq, watermark,
		lease_owner, lease_until, attempt_count, last_error, full_scan_completed_at, updated_at
		FROM tenant_storage_tier_scan WHERE tenant_id = ?`, tenantID).Scan(
		&job.Status, &job.Phase, &job.Cursor, &job.LayerPathHash, &job.LayerEntrySeq, &job.Watermark,
		&job.LeaseOwner, &leaseUntil, &job.AttemptCount, &lastError, &completedAt, &job.UpdatedAt)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, ErrNotFound
	}
	if err != nil {
		return nil, err
	}
	job.LeaseUntil = ptrFromNullTime(leaseUntil)
	job.FullScanCompletedAt = ptrFromNullTime(completedAt)
	if lastError.Valid {
		job.LastError = lastError.String
	}
	return job, nil
}

// ListIncompleteStorageTierScanTenants returns scan jobs that still need a
// worker signal. The query is central-DB-only; callers must retain the
// warm-only tenant acquisition policy when dispatching the signal.
func (s *Store) ListIncompleteStorageTierScanTenants(ctx context.Context, limit int) ([]string, error) {
	if limit <= 0 {
		limit = 1000
	}
	rows, err := s.db.QueryContext(ctx, `SELECT tenant_id
		FROM tenant_storage_tier_scan
		WHERE full_scan_completed_at IS NULL AND status <> 'blocked'
		ORDER BY updated_at ASC LIMIT ?`, limit)
	if err != nil {
		return nil, fmt.Errorf("list incomplete storage tier scans: %w", err)
	}
	defer func() { _ = rows.Close() }()
	var tenants []string
	for rows.Next() {
		var tenantID string
		if err := rows.Scan(&tenantID); err != nil {
			return nil, err
		}
		tenants = append(tenants, tenantID)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return tenants, nil
}

func (s *Store) ClaimStorageTierScanJob(ctx context.Context, tenantID, owner string, leaseDuration time.Duration) (*StorageTierScanJob, bool, error) {
	if leaseDuration <= 0 {
		leaseDuration = 30 * time.Second
	}
	now := time.Now().UTC()
	leaseUntil := now.Add(leaseDuration)
	res, err := s.db.ExecContext(ctx, `UPDATE tenant_storage_tier_scan
		SET status = 'running', lease_owner = ?, lease_until = ?, attempt_count = attempt_count + 1
		WHERE tenant_id = ? AND full_scan_completed_at IS NULL
		  AND (status = 'pending' OR (status = 'running' AND (lease_until IS NULL OR lease_until < ? OR lease_owner = ?)))`,
		owner, leaseUntil, tenantID, now, owner)
	if err != nil {
		return nil, false, err
	}
	n, err := res.RowsAffected()
	if err != nil || n == 0 {
		return nil, false, err
	}
	job, err := s.GetStorageTierScanJob(ctx, tenantID)
	return job, err == nil, err
}

func (s *Store) StorageTierScanWatermarkState(ctx context.Context, tenantID string, watermark int64) (StorageTierMutationState, error) {
	var state StorageTierMutationState
	err := s.db.QueryRowContext(ctx, `SELECT
		COALESCE(SUM(status = 'pending'), 0),
		COALESCE(SUM(status = 'failed'), 0)
		FROM quota_mutation_log
		WHERE tenant_id = ? AND id <= ?
		  AND mutation_type IN ('file_create', 'file_overwrite', 'file_delete', 'upload_complete', 'fs_layer_store')`,
		tenantID, watermark).Scan(&state.Pending, &state.Failed)
	return state, err
}

func (s *Store) InitializeStorageTierCounters(ctx context.Context, tenantID, owner string) error {
	return s.InTx(ctx, func(tx *sql.Tx) error {
		var leaseOwner string
		if err := tx.QueryRow(`SELECT lease_owner FROM tenant_storage_tier_scan WHERE tenant_id = ? FOR UPDATE`, tenantID).Scan(&leaseOwner); err != nil {
			return err
		}
		if leaseOwner != owner {
			return fmt.Errorf("storage tier scan lease lost for tenant %s", tenantID)
		}
		_, err := tx.Exec(`UPDATE tenant_quota_usage q SET
			db_storage_bytes = COALESCE(db_storage_bytes,
				(SELECT COALESCE(SUM(size_bytes), 0) FROM tenant_file_meta WHERE tenant_id = ? AND storage_type = 'db9' AND layer_owned = 0) +
				(SELECT COALESCE(SUM(size_bytes), 0) FROM tenant_layer_storage_meta WHERE tenant_id = ? AND storage_type = 'db9')),
			s3_storage_bytes = COALESCE(s3_storage_bytes,
				(SELECT COALESCE(SUM(size_bytes), 0) FROM tenant_file_meta WHERE tenant_id = ? AND storage_type = 's3' AND layer_owned = 0) +
				(SELECT COALESCE(SUM(size_bytes), 0) FROM tenant_layer_storage_meta WHERE tenant_id = ? AND storage_type = 's3'))
			WHERE q.tenant_id = ?`, tenantID, tenantID, tenantID, tenantID, tenantID)
		return err
	})
}

// ApplyStorageTierScanPage classifies only previously-unclassified central
// rows. Existing typed rows are left untouched, making retries idempotent and
// allowing normal write mutations to race safely with the scan.
func (s *Store) ApplyStorageTierScanPage(ctx context.Context, tenantID, owner, nextCursor string, files []StorageTierScanFile) error {
	return s.InTx(ctx, func(tx *sql.Tx) error {
		var leaseOwner string
		if err := tx.QueryRow(`SELECT lease_owner FROM tenant_storage_tier_scan WHERE tenant_id = ? FOR UPDATE`, tenantID).Scan(&leaseOwner); err != nil {
			return err
		}
		if leaseOwner != owner {
			return fmt.Errorf("storage tier scan lease lost for tenant %s", tenantID)
		}
		for _, file := range files {
			var size int64
			var current sql.NullString
			scanErr := tx.QueryRow(`SELECT size_bytes, storage_type FROM tenant_file_meta WHERE tenant_id = ? AND file_id = ? FOR UPDATE`, tenantID, file.FileID).Scan(&size, &current)
			switch {
			case errors.Is(scanErr, sql.ErrNoRows):
				// The tenant listing is the authoritative source for a missing
				// shadow. Backfill only the tier fields needed for billing; total
				// and file/media counters are owned by the mutation ledger and are
				// deliberately not changed here.
				if file.StorageType != StorageTypeDB9 && file.StorageType != StorageTypeS3 {
					continue
				}
				if _, err := tx.Exec(`INSERT INTO tenant_file_meta
					(tenant_id, file_id, size_bytes, is_media, storage_type, layer_owned)
					VALUES (?, ?, ?, ?, ?, ?)`, tenantID, file.FileID, file.SizeBytes,
					boolToInt(file.IsMedia), file.StorageType, boolToInt(file.LayerOwned)); err != nil {
					return err
				}
				size = file.SizeBytes
				current.Valid = false
			case scanErr != nil:
				return scanErr
			case current.Valid:
				continue
			case file.StorageType != StorageTypeDB9 && file.StorageType != StorageTypeS3:
				// Preserve NULL for future/unknown tiers so the final unclassified
				// check keeps split storage metrics disabled.
				continue
			default:
				res, err := tx.Exec(`UPDATE tenant_file_meta SET storage_type = ?, layer_owned = ? WHERE tenant_id = ? AND file_id = ? AND storage_type IS NULL`, file.StorageType, boolToInt(file.LayerOwned), tenantID, file.FileID)
				if err != nil {
					return err
				}
				changed, err := res.RowsAffected()
				if err != nil || changed == 0 {
					continue
				}
			}
			// Layer-owned core rows are aliases: their physical bytes belong to
			// tenant_layer_storage_meta. The layer phase classifies or backfills
			// that payload, so the core phase must not change total or tier bytes.
			if !file.LayerOwned {
				column := "db_storage_bytes"
				if file.StorageType == StorageTypeS3 {
					column = "s3_storage_bytes"
				}
				if _, err := tx.Exec(`UPDATE tenant_quota_usage SET `+column+` = `+column+` + ? WHERE tenant_id = ?`, size, tenantID); err != nil {
					return err
				}
			}
		}
		if nextCursor != "" {
			_, err := tx.Exec(`UPDATE tenant_storage_tier_scan SET scan_cursor = ?, status = 'pending', lease_owner = '', lease_until = NULL, last_error = NULL WHERE tenant_id = ?`, nextCursor, tenantID)
			return err
		}
		_, err := tx.Exec(`UPDATE tenant_storage_tier_scan SET scan_phase = 'layer', scan_cursor = '',
			layer_path_hash = '', layer_entry_seq = 0, status = 'pending', lease_owner = '', lease_until = NULL,
			last_error = NULL WHERE tenant_id = ?`, tenantID)
		return err
	})
}

// ApplyStorageTierLayerScanPage records immutable fs-layer payload shadows.
// Unlike core rows, a missing layer shadow is safe to create because the
// tenant entry is append-only and represents physical storage still retained.
func (s *Store) ApplyStorageTierLayerScanPage(ctx context.Context, tenantID, owner string, nextCursor StorageTierLayerCursor, layers []StorageTierScanLayer) (completed bool, err error) {
	err = s.InTx(ctx, func(tx *sql.Tx) error {
		var leaseOwner string
		if err := tx.QueryRow(`SELECT lease_owner FROM tenant_storage_tier_scan WHERE tenant_id = ? AND scan_phase = 'layer' FOR UPDATE`, tenantID).Scan(&leaseOwner); err != nil {
			return err
		}
		if leaseOwner != owner {
			return fmt.Errorf("storage tier scan lease lost for tenant %s", tenantID)
		}
		for _, layer := range layers {
			var storageType *StorageType
			if layer.StorageType == StorageTypeDB9 || layer.StorageType == StorageTypeS3 {
				value := layer.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 (?, ?, ?, ?, ?, ?)`,
				tenantID, layer.LayerID, layer.PathHash, layer.EntrySeq, layer.SizeBytes, nullableStorageType(storageType))
			if err != nil {
				return err
			}
			inserted, err := res.RowsAffected()
			if err != nil || inserted == 0 {
				continue
			}
			dbDelta, s3Delta := int64(0), int64(0)
			switch layer.StorageType {
			case StorageTypeDB9:
				dbDelta = layer.SizeBytes
			case StorageTypeS3:
				s3Delta = layer.SizeBytes
			}
			if err := s.IncrQuotaUsageCountersTx(tx, tenantID, layer.SizeBytes, dbDelta, s3Delta, 0, 0, 0); err != nil {
				return err
			}
		}
		if nextCursor != (StorageTierLayerCursor{}) {
			_, err := tx.Exec(`UPDATE tenant_storage_tier_scan SET scan_cursor = ?, layer_path_hash = ?, layer_entry_seq = ?,
				status = 'pending', lease_owner = '', lease_until = NULL, last_error = NULL WHERE tenant_id = ?`,
				nextCursor.LayerID, nextCursor.PathHash, nextCursor.EntrySeq, tenantID)
			return err
		}

		var mutationState StorageTierMutationState
		if err := tx.QueryRow(`SELECT
			COALESCE(SUM(status = 'pending'), 0),
			COALESCE(SUM(status = 'failed'), 0)
			FROM quota_mutation_log WHERE tenant_id = ?
			  AND mutation_type IN ('file_create', 'file_overwrite', 'file_delete', 'upload_complete', 'fs_layer_store')`,
			tenantID).Scan(&mutationState.Pending, &mutationState.Failed); err != nil {
			return err
		}
		var unclassified int64
		if err := tx.QueryRow(`SELECT
			(SELECT COUNT(*) FROM tenant_file_meta WHERE tenant_id = ? AND storage_type IS NULL) +
			(SELECT COUNT(*) FROM tenant_layer_storage_meta WHERE tenant_id = ? AND storage_type IS NULL)`, tenantID, tenantID).Scan(&unclassified); err != nil {
			return err
		}
		var total int64
		var dbBytes, s3Bytes sql.NullInt64
		if err := tx.QueryRow(`SELECT storage_bytes, db_storage_bytes, s3_storage_bytes FROM tenant_quota_usage WHERE tenant_id = ? FOR UPDATE`, tenantID).Scan(&total, &dbBytes, &s3Bytes); err != nil {
			return err
		}
		ready := mutationState.Pending == 0 && mutationState.Failed == 0 && unclassified == 0 && dbBytes.Valid && s3Bytes.Valid && dbBytes.Int64 >= 0 && s3Bytes.Int64 >= 0 && dbBytes.Int64+s3Bytes.Int64 == total
		if ready {
			now := time.Now().UTC()
			_, err := tx.Exec(`UPDATE tenant_storage_tier_scan SET status = 'completed', scan_cursor = '', layer_path_hash = '', layer_entry_seq = 0, lease_owner = '', lease_until = NULL, last_error = NULL, full_scan_completed_at = ? WHERE tenant_id = ?`, now, tenantID)
			completed = err == nil
			return err
		}
		status := StorageTierScanStatusBlocked
		reason := "storage tier readiness invariant failed"
		if mutationState.Pending > 0 && mutationState.Failed == 0 {
			status = StorageTierScanStatusPending
			reason = "storage tier mutations pending"
		} else if mutationState.Failed > 0 {
			reason = "storage tier mutation failed"
		}
		_, err := tx.Exec(`UPDATE tenant_storage_tier_scan SET status = ?, lease_owner = '', lease_until = NULL, last_error = ? WHERE tenant_id = ?`, status, reason, tenantID)
		return err
	})
	return completed, err
}

func (s *Store) BlockStorageTierScanJob(ctx context.Context, tenantID, owner, reason string) error {
	if len(reason) > 2048 {
		reason = reason[:2048]
	}
	return s.InTx(ctx, func(tx *sql.Tx) error {
		var leaseOwner string
		if err := tx.QueryRow(`SELECT lease_owner FROM tenant_storage_tier_scan
			WHERE tenant_id = ? FOR UPDATE`, tenantID).Scan(&leaseOwner); err != nil {
			return err
		}
		if leaseOwner != owner {
			return nil
		}

		// Lock one current failure so a concurrent repair cannot mark the last
		// failed mutation applied between this check and the blocked transition.
		// If the failure was already repaired, restart from core instead of
		// persisting a block based on the scanner's earlier stale observation.
		var failedMutationID int64
		err := tx.QueryRow(`SELECT id FROM quota_mutation_log
			WHERE tenant_id = ? AND status = 'failed'
			  AND mutation_type IN ('file_create', 'file_overwrite', 'file_delete', 'upload_complete', 'fs_layer_store')
			ORDER BY id LIMIT 1 FOR UPDATE`, tenantID).Scan(&failedMutationID)
		switch {
		case err == nil:
			_, err = tx.Exec(`UPDATE tenant_storage_tier_scan
				SET status = 'blocked', lease_owner = '', lease_until = NULL, last_error = ?
				WHERE tenant_id = ? AND lease_owner = ?`, reason, tenantID, owner)
			return err
		case !errors.Is(err, sql.ErrNoRows):
			return err
		}

		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 completed while scan was running; rescan required',
			    s.full_scan_completed_at = NULL
			WHERE s.tenant_id = ? AND s.lease_owner = ?`, tenantID, owner)
		if err != nil {
			return err
		}
		updated, err := res.RowsAffected()
		if err != nil || updated == 0 {
			return err
		}
		_, err = tx.Exec(`INSERT INTO tenant_notify_outbox (tenant_id, work_mask) VALUES (?, ?)`,
			tenantID, TenantNotifyWorkStorageTierScan)
		return err
	})
}

func (s *Store) ReleaseStorageTierScanJob(ctx context.Context, tenantID, owner string, scanErr error) error {
	message := ""
	if scanErr != nil {
		message = scanErr.Error()
		if len(message) > 2048 {
			message = message[:2048]
		}
	}
	_, err := s.db.ExecContext(ctx, `UPDATE tenant_storage_tier_scan SET status = 'pending', lease_owner = '', lease_until = NULL, last_error = ? WHERE tenant_id = ? AND lease_owner = ?`, message, tenantID, owner)
	return err
}