package content

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

const (
	MaxGCLeaseBatch = 100
	MaxGCAttempts   = 10

	GCActionStale    = "stale"
	GCActionRetained = "retained"
	GCActionDelete   = "delete"
)

// GCBindingScope constrains one cleanup owner to one registered product
// binding. It is intentionally exact: no wildcard or workspace-wide mode.
type GCBindingScope struct {
	WorkspaceID string
	Domain      string
	BindingKey  string
}

func normalizeGCBindingScope(scope GCBindingScope) (GCBindingScope, error) {
	scope.WorkspaceID = strings.TrimSpace(scope.WorkspaceID)
	scope.Domain = strings.TrimSpace(scope.Domain)
	scope.BindingKey = strings.TrimSpace(scope.BindingKey)
	if scope.WorkspaceID == "" || scope.Domain == "" || scope.BindingKey == "" ||
		scope.Domain == AuthorityDomainDefaultBinding || scope.BindingKey == AuthorityDomainDefaultBinding {
		return GCBindingScope{}, fmt.Errorf("content gc binding scope must be exact")
	}
	return scope, nil
}

// LeaseDueStagedPayloadGCForBinding claims one bounded batch belonging to one
// exact content binding. The join prevents this owner from leasing unrelated
// workspace/domain work; PrepareStagedPayloadGC still installs the deletion
// fence before payload deletion.
func (s *Store) LeaseDueStagedPayloadGCForBinding(ctx context.Context, scope GCBindingScope, now time.Time, leaseOwner string, leaseUntil time.Time, limit int) ([]StagedPayloadGC, error) {
	var err error
	if scope, err = normalizeGCBindingScope(scope); err != nil {
		return nil, err
	}
	if err := s.ensureDB(); err != nil {
		return nil, err
	}
	leaseOwner = strings.TrimSpace(leaseOwner)
	if leaseOwner == "" || limit <= 0 || limit > MaxGCLeaseBatch || !leaseUntil.After(now) {
		return nil, fmt.Errorf("content gc lease request is invalid")
	}
	var leased []StagedPayloadGC
	err = s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		var candidates []StagedPayloadGC
		query := tx.Table("content_staged_payload_gc AS gc").
			Select("gc.id", "gc.workspace_id", "gc.content_id", "gc.version_id", "gc.model", "gc.payload_ref", "gc.status", "gc.attempts", "gc.lease_until").
			Joins("JOIN versioned_content AS content ON content.workspace_id = gc.workspace_id AND content.id = gc.content_id").
			Where("content.workspace_id = ? AND content.domain = ? AND content.binding_key = ?", scope.WorkspaceID, scope.Domain, scope.BindingKey).
			Where("gc.status IN ? AND gc.delete_after <= ? AND (gc.lease_until IS NULL OR gc.lease_until <= ?)", []string{GCStatusPending, GCStatusDeleting}, now, now).
			Order("gc.delete_after ASC, gc.id ASC")
		if err := query.Limit(limit).Find(&candidates).Error; err != nil {
			return err
		}
		for _, candidate := range candidates {
			claim := tx.Model(&StagedPayloadGC{}).
				Where("id = ? AND status = ? AND delete_after <= ? AND (lease_until IS NULL OR lease_until <= ?)", candidate.ID, candidate.Status, now, now)
			claim = claim.Where("workspace_id = ? AND content_id = ?", scope.WorkspaceID, candidate.ContentID)
			result := claim.Updates(map[string]any{"lease_owner": leaseOwner, "lease_until": leaseUntil, "updated_at": now})
			if result.Error != nil {
				return result.Error
			}
			if result.RowsAffected == 1 {
				candidate.LeaseOwner = leaseOwner
				candidate.LeaseUntil = &leaseUntil
				candidate.UpdatedAt = now
				leased = append(leased, candidate)
			}
		}
		return nil
	})
	if err != nil {
		return nil, fmt.Errorf("lease staged content payload gc: %w", err)
	}
	return leased, nil
}

// PrepareStagedPayloadGC atomically reconciles exact version reachability and
// changes an unreachable row to deleting. PublishVersion can retain only a
// pending row, so once this fence commits no version can begin referencing the
// payload being deleted.
func (s *Store) PrepareStagedPayloadGC(ctx context.Context, row StagedPayloadGC, leaseOwner string, now time.Time) (string, error) {
	if err := s.ensureDB(); err != nil {
		return "", err
	}
	if row.LeaseUntil == nil {
		return GCActionStale, nil
	}
	action := GCActionStale
	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		var locked StagedPayloadGC
		err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
			Where("id = ? AND status IN ? AND lease_owner = ? AND lease_until = ?", strings.TrimSpace(row.ID), []string{GCStatusPending, GCStatusDeleting}, strings.TrimSpace(leaseOwner), *row.LeaseUntil).
			First(&locked).Error
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil
		}
		if err != nil {
			return err
		}
		var reachable int64
		if err := tx.Model(&ContentVersion{}).
			Where("workspace_id = ? AND content_id = ? AND id = ?", locked.WorkspaceID, locked.ContentID, locked.VersionID).
			Limit(1).Count(&reachable).Error; err != nil {
			return err
		}
		if reachable != 0 {
			result := tx.Model(&StagedPayloadGC{}).
				Where("id = ? AND status IN ? AND lease_owner = ? AND lease_until = ?", locked.ID, []string{GCStatusPending, GCStatusDeleting}, strings.TrimSpace(leaseOwner), *row.LeaseUntil).
				Updates(map[string]any{"status": GCStatusRetained, "lease_owner": "", "lease_until": nil, "last_error": "", "completed_at": now, "updated_at": now})
			if result.Error != nil {
				return result.Error
			}
			if result.RowsAffected == 1 {
				action = GCActionRetained
			}
			return nil
		}
		result := tx.Model(&StagedPayloadGC{}).
			Where("id = ? AND status IN ? AND lease_owner = ? AND lease_until = ?", locked.ID, []string{GCStatusPending, GCStatusDeleting}, strings.TrimSpace(leaseOwner), *row.LeaseUntil).
			Updates(map[string]any{"status": GCStatusDeleting, "updated_at": now})
		if result.Error != nil {
			return result.Error
		}
		if result.RowsAffected == 1 {
			action = GCActionDelete
		}
		return nil
	})
	if err != nil {
		return "", fmt.Errorf("prepare staged content payload gc: %w", err)
	}
	return action, nil
}

func (s *Store) MarkStagedPayloadGCDeleted(ctx context.Context, row StagedPayloadGC, leaseOwner string, now time.Time) (bool, error) {
	if err := s.ensureDB(); err != nil {
		return false, err
	}
	if row.LeaseUntil == nil {
		return false, nil
	}
	result := s.db.WithContext(ctx).Model(&StagedPayloadGC{}).
		Where("id = ? AND status = ? AND lease_owner = ? AND lease_until = ?", strings.TrimSpace(row.ID), GCStatusDeleting, strings.TrimSpace(leaseOwner), *row.LeaseUntil).
		Updates(map[string]any{"status": GCStatusDeleted, "lease_owner": "", "lease_until": nil, "last_error": "", "completed_at": now, "updated_at": now})
	if result.Error != nil {
		return false, fmt.Errorf("mark staged content payload deleted: %w", result.Error)
	}
	return result.RowsAffected == 1, nil
}

// MarkStagedPayloadGCAttemptFailed releases one deletion lease and schedules a
// bounded retry, or terminally fails after MaxGCAttempts.
func (s *Store) MarkStagedPayloadGCAttemptFailed(ctx context.Context, row StagedPayloadGC, leaseOwner string, failure error, now time.Time) (bool, error) {
	if err := s.ensureDB(); err != nil {
		return false, err
	}
	if row.LeaseUntil == nil {
		return false, nil
	}
	attempts := row.Attempts + 1
	status := GCStatusPending
	nextAttempt := now.Add(deliveryRetryBackoff(attempts))
	var completedAt *time.Time
	if attempts >= MaxGCAttempts {
		status = GCStatusFailed
		nextAttempt = now
		completedAt = &now
	}
	lastError := ""
	if failure != nil {
		lastError = truncateDeliveryError(failure.Error())
	}
	result := s.db.WithContext(ctx).Model(&StagedPayloadGC{}).
		Where("id = ? AND status = ? AND lease_owner = ? AND lease_until = ?", strings.TrimSpace(row.ID), GCStatusDeleting, strings.TrimSpace(leaseOwner), *row.LeaseUntil).
		Updates(map[string]any{
			"status": status, "attempts": attempts, "delete_after": nextAttempt,
			"lease_owner": "", "lease_until": nil, "last_error": lastError,
			"completed_at": completedAt, "updated_at": now,
		})
	if result.Error != nil {
		return false, fmt.Errorf("mark staged content payload gc attempt failed: %w", result.Error)
	}
	return result.RowsAffected == 1, nil
}