package datastore

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

	"github.com/google/uuid"
)

// File GC is a correctness cleanup path, not best-effort enrichment work.
// A zero max_attempts means retry indefinitely until the blob/quota cleanup is
// done or an operator explicitly updates the task policy.
const defaultFileGCMaxAttempts = 0

type FileGCTaskStatus string

const (
	FileGCTaskQueued       FileGCTaskStatus = "queued"
	FileGCTaskProcessing   FileGCTaskStatus = "processing"
	FileGCTaskSucceeded    FileGCTaskStatus = "succeeded"
	FileGCTaskDeadLettered FileGCTaskStatus = "dead_lettered"
)

// FileGCTask records durable cleanup for an orphaned file identity. The task is
// intentionally keyed by file_id/storage_ref rather than path so retries cannot
// delete another actor's later recreate at the same path. It does not track old
// blob refs produced by overwriting a still-live file_id; those need a separate
// blob-ref GC path.
type FileGCTask struct {
	TaskID        string
	FileID        string
	StorageType   StorageType
	ContentLayout ContentLayout
	StorageRef    string
	SizeBytes     int64
	ContentType   string
	Status        FileGCTaskStatus
	AttemptCount  int
	MaxAttempts   int
	Receipt       string
	LeasedAt      *time.Time
	LeaseUntil    *time.Time
	AvailableAt   time.Time
	LastError     string
	CreatedAt     time.Time
	UpdatedAt     time.Time
	CompletedAt   *time.Time
}

// EnqueueFileGCTaskTx inserts a queued file cleanup task inside an existing
// metadata transaction. Duplicate tasks for the same file_id are treated as a
// successful no-op because file IDs are never reused.
func (s *Store) EnqueueFileGCTaskTx(db execer, task *FileGCTask) (bool, error) {
	return s.enqueueFileGCTask(db, task)
}

// GetFileGCTaskByFileID returns the durable cleanup task for a file.
func (s *Store) GetFileGCTaskByFileID(ctx context.Context, fileID string) (out *FileGCTask, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "get_file_gc_task_by_file_id", start, &err)

	row := s.db.QueryRowContext(ctx, `SELECT task_id, file_id, storage_type, content_layout, storage_ref,
		size_bytes, content_type, status, attempt_count, max_attempts, receipt,
		leased_at, lease_until, available_at, last_error, created_at, updated_at, completed_at
		FROM file_gc_tasks WHERE `+s.scope.And(`file_id = ?`), s.scope.Args(fileID)...)
	return scanFileGCTask(row)
}

func (s *Store) ListFileGCTaskS3Refs(ctx context.Context, cursor string, limit int) ([]ConfirmedS3Ref, string, error) {
	if limit <= 0 {
		limit = 500
	}
	var rows *sql.Rows
	var err error
	if cursor == "" {
		rows, err = s.db.QueryContext(ctx,
			`SELECT DISTINCT storage_ref, content_layout
			 FROM file_gc_tasks
			 WHERE `+s.scope.And(`storage_type = 's3' AND storage_ref <> ''`)+`
			 ORDER BY storage_ref ASC LIMIT ?`, s.scope.Args(limit)...)
	} else {
		rows, err = s.db.QueryContext(ctx,
			`SELECT DISTINCT storage_ref, content_layout
			 FROM file_gc_tasks
			 WHERE `+s.scope.And(`storage_type = 's3' AND storage_ref <> '' AND storage_ref > ?`)+`
			 ORDER BY storage_ref ASC LIMIT ?`, s.scope.Args(cursor, limit)...)
	}
	if err != nil {
		return nil, "", fmt.Errorf("query file gc s3 refs: %w", err)
	}
	defer func() { _ = rows.Close() }()

	out := make([]ConfirmedS3Ref, 0)
	for rows.Next() {
		var ref ConfirmedS3Ref
		if err := rows.Scan(&ref.StorageRef, &ref.ContentLayout); err != nil {
			return nil, "", fmt.Errorf("scan file gc s3 ref: %w", err)
		}
		ref.StorageRefHash = StorageRefHash(ref.StorageRef)
		out = append(out, ref)
	}
	if err := rows.Err(); err != nil {
		return nil, "", err
	}
	nextCursor := ""
	if len(out) == limit {
		nextCursor = out[len(out)-1].StorageRef
	}
	return out, nextCursor, nil
}

// CountQueuedFileGCTasks returns the number of file_gc tasks in queued status.
// This is a read-only query used by the safety-net scan to detect unclaimed
// tasks without claiming them (unlike ClaimFileGCTask which mutates state).
// Includes tasks with future available_at (retry backoff); for claim-parity
// diagnostics prefer CountClaimableFileGCTasks.
func (s *Store) CountQueuedFileGCTasks(ctx context.Context) (count int64, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "count_queued_file_gc_tasks", start, &err)
	err = s.db.QueryRowContext(ctx,
		`SELECT COUNT(*) FROM file_gc_tasks WHERE `+s.scope.And(`status = ?`),
		s.scope.Args(FileGCTaskQueued)...).Scan(&count)
	return count, err
}

// CountClaimableFileGCTasks returns queued tasks whose available_at is due,
// matching ClaimFileGCTask's predicate. Used to detect "claim empty but work
// is visible" without treating in-backoff retries as a stuck backlog.
func (s *Store) CountClaimableFileGCTasks(ctx context.Context, now time.Time) (count int64, err error) {
	count, _, err = s.ObserveClaimableFileGCTasks(ctx, now)
	return count, err
}

// ObserveClaimableFileGCTasks returns the claimable queued count and the oldest
// available_at among those rows (claim parity with ClaimFileGCTask). When
// count is 0, oldest is the zero time.
func (s *Store) ObserveClaimableFileGCTasks(ctx context.Context, now time.Time) (count int64, oldest time.Time, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "observe_claimable_file_gc_tasks", start, &err)
	if now.IsZero() {
		now = time.Now().UTC()
	} else {
		now = now.UTC()
	}
	var oldestSQL sql.NullTime
	err = s.db.QueryRowContext(ctx,
		`SELECT COUNT(*), MIN(available_at) FROM file_gc_tasks WHERE `+s.scope.And(`status = ? AND available_at <= ?`),
		s.scope.Args(FileGCTaskQueued, now)...).Scan(&count, &oldestSQL)
	if err != nil {
		return 0, time.Time{}, err
	}
	if oldestSQL.Valid {
		oldest = oldestSQL.Time.UTC()
	}
	return count, oldest, nil
}

// ClaimFileGCTask claims one queued file GC task and leases it to the caller.
func (s *Store) ClaimFileGCTask(ctx context.Context, now time.Time, leaseDuration time.Duration) (out *FileGCTask, found bool, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "claim_file_gc_task", start, &err)

	if now.IsZero() {
		now = time.Now().UTC()
	} else {
		now = now.UTC()
	}
	if leaseDuration <= 0 {
		leaseDuration = 30 * time.Second
	}

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, false, err
	}
	defer func() { _ = tx.Rollback() }()

	// SelCols adds the fs_id projection in shared shape to work around the
	// TiDB FOR UPDATE SKIP LOCKED planner bug that rejects WHERE predicates
	// over unprojected columns (see Scope.SelCols).
	row := tx.QueryRowContext(ctx, `SELECT `+s.scope.SelCols(`task_id, file_id, storage_type, content_layout, storage_ref,
		size_bytes, content_type, status, attempt_count, max_attempts, receipt,
		leased_at, lease_until, available_at, last_error, created_at, updated_at, completed_at`)+`
		FROM file_gc_tasks
		WHERE `+s.scope.And(`status = ? AND available_at <= ?`)+`
		ORDER BY available_at, created_at, task_id
		LIMIT 1
		FOR UPDATE SKIP LOCKED`, s.scope.Args(FileGCTaskQueued, now)...)
	// The claim query projects fs_id first in shared shape (workaround for
	// the TiDB SKIP LOCKED planner bug; see Scope.SelCols); withFsID skips it.
	task, err := scanFileGCTask(row, s.scope.Shared())
	if err != nil {
		if errors.Is(err, ErrNotFound) {
			return nil, false, nil
		}
		return nil, false, err
	}

	receipt := uuid.NewString()
	leasedAt := now
	leaseUntil := now.Add(leaseDuration)
	res, err := tx.ExecContext(ctx, `UPDATE file_gc_tasks SET status = ?,
		attempt_count = attempt_count + 1, receipt = ?, leased_at = ?,
		lease_until = ?, updated_at = ?
		WHERE `+s.scope.And(`task_id = ? AND status = ? AND available_at <= ?`),
		append([]any{FileGCTaskProcessing, receipt, leasedAt, leaseUntil, now},
			s.scope.Args(task.TaskID, FileGCTaskQueued, now)...)...)
	if err != nil {
		return nil, false, err
	}
	rowsAffected, err := res.RowsAffected()
	if err != nil {
		return nil, false, err
	}
	if rowsAffected == 0 {
		return nil, false, nil
	}

	task.Status = FileGCTaskProcessing
	task.AttemptCount++
	task.Receipt = receipt
	task.LeasedAt = &leasedAt
	task.LeaseUntil = &leaseUntil
	task.UpdatedAt = now

	if err := tx.Commit(); err != nil {
		return nil, false, err
	}
	return task, true, nil
}

// AckFileGCTask marks a leased file GC task as succeeded.
func (s *Store) AckFileGCTask(ctx context.Context, taskID, receipt string) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "ack_file_gc_task", start, &err)

	now := time.Now().UTC()
	res, err := s.db.ExecContext(ctx, `UPDATE file_gc_tasks SET status = ?, receipt = NULL,
		leased_at = NULL, lease_until = NULL, completed_at = ?, updated_at = ?
		WHERE `+s.scope.And(`task_id = ? AND status = ? AND receipt = ? AND lease_until IS NOT NULL AND lease_until > ?`),
		append([]any{FileGCTaskSucceeded, now, now},
			s.scope.Args(taskID, FileGCTaskProcessing, receipt, now)...)...)
	if err != nil {
		return err
	}
	rowsAffected, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if rowsAffected > 0 {
		return nil
	}
	return s.fileGCTaskLeaseError(ctx, taskID)
}

// RetryFileGCTask requeues or dead-letters a leased file GC task.
func (s *Store) RetryFileGCTask(ctx context.Context, taskID, receipt string, retryAt time.Time, lastErr string) (err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "retry_file_gc_task", start, &err)

	now := time.Now().UTC()
	if retryAt.IsZero() {
		retryAt = now
	} else {
		retryAt = retryAt.UTC()
	}

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	row := tx.QueryRowContext(ctx, `SELECT task_id, file_id, storage_type, content_layout, storage_ref,
		size_bytes, content_type, status, attempt_count, max_attempts, receipt,
		leased_at, lease_until, available_at, last_error, created_at, updated_at, completed_at
		FROM file_gc_tasks WHERE `+s.scope.And(`task_id = ?`)+` FOR UPDATE`, s.scope.Args(taskID)...)
	task, err := scanFileGCTask(row)
	if err != nil {
		return err
	}
	if task.Status != FileGCTaskProcessing || task.Receipt != receipt || task.LeaseUntil == nil || !task.LeaseUntil.After(now) {
		return ErrFileGCTaskLeaseMismatch
	}

	if task.MaxAttempts > 0 && task.AttemptCount >= task.MaxAttempts {
		_, err = tx.ExecContext(ctx, `UPDATE file_gc_tasks SET status = ?, receipt = NULL,
			leased_at = NULL, lease_until = NULL, last_error = ?, completed_at = ?, updated_at = ?
			WHERE `+s.scope.And(`task_id = ?`),
			append([]any{FileGCTaskDeadLettered, nullStr(lastErr), now, now}, s.scope.Args(taskID)...)...)
	} else {
		_, err = tx.ExecContext(ctx, `UPDATE file_gc_tasks SET status = ?, receipt = NULL,
			leased_at = NULL, lease_until = NULL, available_at = ?, last_error = ?,
			completed_at = NULL, updated_at = ?
			WHERE `+s.scope.And(`task_id = ?`),
			append([]any{FileGCTaskQueued, retryAt, nullStr(lastErr), now}, s.scope.Args(taskID)...)...)
	}
	if err != nil {
		return err
	}
	return tx.Commit()
}

// RecoverExpiredFileGCTasks requeues processing tasks whose lease expired.
func (s *Store) RecoverExpiredFileGCTasks(ctx context.Context, now time.Time, limit int) (recovered int, err error) {
	start := time.Now()
	defer observeStoreOp(ctx, "recover_expired_file_gc_tasks", start, &err)

	if now.IsZero() {
		now = time.Now().UTC()
	} else {
		now = now.UTC()
	}

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return 0, err
	}
	defer func() { _ = tx.Rollback() }()

	query := `SELECT ` + s.scope.SelCols(`task_id`) + ` FROM file_gc_tasks
		WHERE ` + s.scope.And(`status = ? AND lease_until IS NOT NULL AND lease_until < ?`) + `
		ORDER BY lease_until, created_at, task_id`
	args := s.scope.Args(FileGCTaskProcessing, now)
	if limit > 0 {
		query += ` LIMIT ?`
		args = append(args, limit)
	}
	query += ` FOR UPDATE SKIP LOCKED`
	rows, err := tx.QueryContext(ctx, query, args...)
	if err != nil {
		return 0, err
	}
	defer func() { _ = rows.Close() }()

	var taskIDs []string
	for rows.Next() {
		// Shared shape projects fs_id first (workaround for the TiDB SKIP
		// LOCKED planner bug; see Scope.SelCols); scan and discard it.
		var fsID int64
		var taskID string
		if s.scope.Shared() {
			if err := rows.Scan(&fsID, &taskID); err != nil {
				return 0, err
			}
		} else if err := rows.Scan(&taskID); err != nil {
			return 0, err
		}
		taskIDs = append(taskIDs, taskID)
	}
	if err := rows.Err(); err != nil {
		return 0, err
	}

	for _, taskID := range taskIDs {
		res, err := tx.ExecContext(ctx, `UPDATE file_gc_tasks SET status = ?, receipt = NULL,
			leased_at = NULL, lease_until = NULL, available_at = ?, updated_at = ?
			WHERE `+s.scope.And(`task_id = ? AND status = ? AND lease_until IS NOT NULL AND lease_until < ?`),
			append([]any{FileGCTaskQueued, now, now},
				s.scope.Args(taskID, FileGCTaskProcessing, now)...)...)
		if err != nil {
			return 0, err
		}
		rowsAffected, err := res.RowsAffected()
		if err != nil {
			return 0, err
		}
		recovered += int(rowsAffected)
	}

	if err := tx.Commit(); err != nil {
		return 0, err
	}
	return recovered, nil
}

func NewFileGCTaskFromFile(f *File, now time.Time) (*FileGCTask, error) {
	if f == nil {
		return nil, fmt.Errorf("file is required")
	}
	if strings.TrimSpace(f.FileID) == "" {
		return nil, fmt.Errorf("file_id is required")
	}
	if now.IsZero() {
		now = time.Now().UTC()
	} else {
		now = now.UTC()
	}
	return &FileGCTask{
		TaskID:        f.FileID,
		FileID:        f.FileID,
		StorageType:   f.StorageType,
		ContentLayout: f.ContentLayout,
		StorageRef:    f.StorageRef,
		SizeBytes:     f.SizeBytes,
		ContentType:   f.ContentType,
		Status:        FileGCTaskQueued,
		MaxAttempts:   defaultFileGCMaxAttempts,
		AvailableAt:   now,
		CreatedAt:     now,
		UpdatedAt:     now,
	}, nil
}

func (s *Store) enqueueFileGCTask(db execer, task *FileGCTask) (bool, error) {
	if task == nil {
		return false, fmt.Errorf("file gc task is required")
	}
	if strings.TrimSpace(task.FileID) == "" {
		return false, fmt.Errorf("file_id is required")
	}
	normalizeFileGCTask(task, time.Now().UTC())
	_, err := db.Exec(`INSERT INTO file_gc_tasks
		(`+s.scope.InsCols(`task_id, file_id, storage_type, content_layout, storage_ref, size_bytes, content_type,
		 status, attempt_count, max_attempts, receipt, leased_at, lease_until,
		 available_at, last_error, created_at, updated_at, completed_at`)+`)
		VALUES (`+s.scope.InsVals(`?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?`)+`)`,
		s.scope.Args(task.TaskID, task.FileID, task.StorageType, contentLayoutForWrite(task.ContentLayout), task.StorageRef, task.SizeBytes,
			nullStr(task.ContentType), task.Status, task.AttemptCount, task.MaxAttempts,
			nullStr(task.Receipt), nilTime(task.LeasedAt), nilTime(task.LeaseUntil),
			task.AvailableAt.UTC(), nullStr(task.LastError), task.CreatedAt.UTC(),
			task.UpdatedAt.UTC(), nilTime(task.CompletedAt))...)
	if err == nil {
		return true, nil
	}
	if isUniqueViolation(err) {
		return false, nil
	}
	return false, err
}

// normalizeFileGCTask applies the defaulting rules of enqueueFileGCTask.
func normalizeFileGCTask(task *FileGCTask, now time.Time) {
	if strings.TrimSpace(task.TaskID) == "" {
		task.TaskID = task.FileID
	}
	if task.Status == "" {
		task.Status = FileGCTaskQueued
	}
	if task.MaxAttempts <= 0 {
		task.MaxAttempts = defaultFileGCMaxAttempts
	}
	if task.AvailableAt.IsZero() {
		task.AvailableAt = now
	}
	if task.CreatedAt.IsZero() {
		task.CreatedAt = now
	}
	if task.UpdatedAt.IsZero() {
		task.UpdatedAt = task.CreatedAt
	}
}

// enqueueFileGCTasksTx inserts many queued tasks in chunked multi-row
// statements, giving the same duplicate-is-no-op semantics as
// enqueueFileGCTask (file IDs are never reused; hardlink batches may repeat a
// file_id). ON DUPLICATE KEY UPDATE (not INSERT IGNORE) is deliberate: only
// unique-key conflicts are swallowed, while real errors (bad types, length,
// constraints) still fail the batch instead of silently dropping GC tasks.
// It returns the number of tasks actually inserted.
func (s *Store) enqueueFileGCTasksTx(db execer, tasks []*FileGCTask) (int64, error) {
	var inserted int64
	now := time.Now().UTC()
	for start := 0; start < len(tasks); start += deleteFileIDBatchSize {
		end := start + deleteFileIDBatchSize
		if end > len(tasks) {
			end = len(tasks)
		}
		batch := tasks[start:end]
		if len(batch) == 0 {
			continue
		}
		values := make([]string, 0, len(batch))
		args := make([]any, 0, len(batch)*18)
		for _, task := range batch {
			if task == nil {
				return inserted, fmt.Errorf("file gc task is required")
			}
			if strings.TrimSpace(task.FileID) == "" {
				return inserted, fmt.Errorf("file_id is required")
			}
			normalizeFileGCTask(task, now)
			values = append(values, `(`+s.scope.InsVals(`?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?`)+`)`)
			args = append(args, s.scope.Args(task.TaskID, task.FileID, task.StorageType, contentLayoutForWrite(task.ContentLayout), task.StorageRef, task.SizeBytes,
				nullStr(task.ContentType), task.Status, task.AttemptCount, task.MaxAttempts,
				nullStr(task.Receipt), nilTime(task.LeasedAt), nilTime(task.LeaseUntil),
				task.AvailableAt.UTC(), nullStr(task.LastError), task.CreatedAt.UTC(),
				task.UpdatedAt.UTC(), nilTime(task.CompletedAt))...)
		}
		res, err := db.Exec(`INSERT INTO file_gc_tasks
		(`+s.scope.InsCols(`task_id, file_id, storage_type, content_layout, storage_ref, size_bytes, content_type,
			 status, attempt_count, max_attempts, receipt, leased_at, lease_until,
			 available_at, last_error, created_at, updated_at, completed_at`)+`)
			VALUES `+strings.Join(values, `,`)+`
			ON DUPLICATE KEY UPDATE task_id = task_id`, args...)
		if err != nil {
			return inserted, err
		}
		n, _ := res.RowsAffected()
		// With the no-op update form (task_id = task_id), MySQL reports 0
		// affected rows for a duplicate key, so n is already <= len(batch).
		// Clamp defensively in case the update clause ever changes to a form
		// that reports changed rows.
		inserted += min(n, int64(len(batch)))
	}
	return inserted, nil
}

func (s *Store) fileGCTaskLeaseError(ctx context.Context, taskID string) error {
	var count int
	err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM file_gc_tasks WHERE `+s.scope.And(`task_id = ?`), s.scope.Args(taskID)...).Scan(&count)
	if err != nil {
		return err
	}
	if count == 0 {
		return ErrNotFound
	}
	return ErrFileGCTaskLeaseMismatch
}

func scanFileGCTask(s scanner, withFsID ...bool) (*FileGCTask, error) {
	var task FileGCTask
	var contentType, receipt, lastError sql.NullString
	var leasedAt, leaseUntil, completedAt sql.NullTime
	// withFsID[0] is set by the SKIP LOCKED claim paths: in shared shape those
	// queries project fs_id first (workaround for the TiDB SKIP LOCKED
	// planner bug; see Scope.SelCols), so scan and discard the leading column.
	dest := make([]any, 0, 18)
	if len(withFsID) > 0 && withFsID[0] {
		var fsID int64
		dest = append(dest, &fsID)
	}
	dest = append(dest, &task.TaskID, &task.FileID, &task.StorageType, &task.ContentLayout, &task.StorageRef,
		&task.SizeBytes, &contentType, &task.Status, &task.AttemptCount,
		&task.MaxAttempts, &receipt, &leasedAt, &leaseUntil, &task.AvailableAt,
		&lastError, &task.CreatedAt, &task.UpdatedAt, &completedAt)
	err := s.Scan(dest...)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	task.ContentType = contentType.String
	task.Receipt = receipt.String
	task.LastError = lastError.String
	task.AvailableAt = task.AvailableAt.UTC()
	task.CreatedAt = task.CreatedAt.UTC()
	task.UpdatedAt = task.UpdatedAt.UTC()
	if leasedAt.Valid {
		t := leasedAt.Time.UTC()
		task.LeasedAt = &t
	}
	if leaseUntil.Valid {
		t := leaseUntil.Time.UTC()
		task.LeaseUntil = &t
	}
	if completedAt.Valid {
		t := completedAt.Time.UTC()
		task.CompletedAt = &t
	}
	return &task, nil
}