files/internal/jobs/pending_upload.go

package jobs

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"git.apinb.com/bsm-sdk/core/logger"
	"git.apinb.com/ops/files/internal/config"
	"git.apinb.com/ops/files/internal/impl"
	"git.apinb.com/ops/files/internal/lifecycle"
	"git.apinb.com/ops/files/internal/models"
	"git.apinb.com/ops/files/internal/storage"
)

// CleanupStatus 描述过期上传清理任务的运行状态。
type CleanupStatus struct {
	Running     bool      `json:"running"`
	LastStarted time.Time `json:"last_started,omitempty"`
	LastSuccess time.Time `json:"last_success,omitempty"`
	LastError   string    `json:"last_error,omitempty"`
}

var cleanupRuntime struct {
	sync.RWMutex
	CleanupStatus
	wg sync.WaitGroup
}

// StartPendingUploadCleanup 启动可取消的过期上传清理任务。
func StartPendingUploadCleanup(ctx context.Context) error {
	cleanupRuntime.Lock()
	if cleanupRuntime.Running {
		cleanupRuntime.Unlock()
		return fmt.Errorf("过期上传清理任务已启动")
	}
	cleanupRuntime.Unlock()

	if err := runCleanup(ctx); err != nil {
		return fmt.Errorf("首次清理过期上传失败: %w", err)
	}

	cleanupRuntime.Lock()
	cleanupRuntime.Running = true
	cleanupRuntime.Unlock()
	cleanupRuntime.wg.Add(1)
	go func() {
		defer cleanupRuntime.wg.Done()
		defer func() {
			cleanupRuntime.Lock()
			cleanupRuntime.Running = false
			cleanupRuntime.Unlock()
		}()

		ticker := time.NewTicker(time.Duration(config.Spec.Cleanup.IntervalSeconds) * time.Second)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				if err := runCleanup(ctx); err != nil && !errors.Is(err, context.Canceled) {
					logger.Errorf("stage=cleanup error=%v", err)
				}
			}
		}
	}()
	return nil
}

// Status 返回清理任务状态快照。
func Status() CleanupStatus {
	cleanupRuntime.RLock()
	defer cleanupRuntime.RUnlock()
	return cleanupRuntime.CleanupStatus
}

// Wait 等待清理任务退出。
func Wait(ctx context.Context) error {
	done := make(chan struct{})
	go func() {
		cleanupRuntime.wg.Wait()
		close(done)
	}()
	select {
	case <-done:
		return waitIntegrity(ctx)
	case <-ctx.Done():
		return ctx.Err()
	}
}

func runCleanup(ctx context.Context) error {
	startedAt := time.Now().UTC()
	cleanupRuntime.Lock()
	cleanupRuntime.LastStarted = startedAt
	cleanupRuntime.Unlock()

	err := cleanupPendingUploads(ctx)
	cleanupRuntime.Lock()
	defer cleanupRuntime.Unlock()
	if err != nil {
		cleanupRuntime.LastError = err.Error()
		return err
	}
	cleanupRuntime.LastSuccess = time.Now().UTC()
	cleanupRuntime.LastError = ""
	return nil
}

func cleanupPendingUploads(ctx context.Context) error {
	var fileObjects []models.FileObject
	now := time.Now()
	result := impl.DBService.WithContext(ctx).
		Where(
			"(status = ? AND expires_at <= ?) OR (status = ? AND (delete_lease_until IS NULL OR delete_lease_until <= ?))",
			models.FileStatusPending,
			now,
			models.FileStatusDeleting,
			now,
		).
		Order("id ASC").
		Limit(100).
		Find(&fileObjects)
	if result.Error != nil {
		return fmt.Errorf("扫描待清理文件失败: %w", result.Error)
	}

	var cleanupErrors []error
	for _, fileObject := range fileObjects {
		var lease lifecycle.DeletionLease
		var claimed bool
		var err error
		now = time.Now()
		if fileObject.Status == models.FileStatusPending {
			lease, claimed, err = lifecycle.ClaimExpiredPendingDeletion(impl.DBService, fileObject.ID, now)
		} else {
			lease, claimed, err = lifecycle.ClaimDeletionRetry(impl.DBService, fileObject.ID, now)
		}
		if err != nil {
			logger.Errorf("identity=%s stage=claim", fileObject.Identity)
			cleanupErrors = append(cleanupErrors, fmt.Errorf("文件 %s 获取删除租约失败: %w", fileObject.Identity, err))
			continue
		}
		if !claimed {
			continue
		}

		if err := cleanupClaimedFile(ctx, fileObject, lease); err != nil {
			cleanupErrors = append(cleanupErrors, err)
		}
	}
	return errors.Join(cleanupErrors...)
}

func cleanupClaimedFile(ctx context.Context, fileObject models.FileObject, lease lifecycle.DeletionLease) error {
	operationCtx, cancel := context.WithTimeout(ctx, lifecycle.DeleteOperationTimeout)
	defer cancel()

	if receiver, ok := impl.StorageService.(storage.UploadReceiver); ok {
		if err := receiver.DiscardUpload(operationCtx, fileObject.Identity); err != nil {
			return fmt.Errorf("文件 %s 清理暂存内容失败: %w", fileObject.Identity, err)
		}
	}
	if err := impl.StorageService.Delete(operationCtx, fileObject.ObjectKey); err != nil {
		logger.Errorf("identity=%s stage=storage_delete", fileObject.Identity)
		return fmt.Errorf("文件 %s 删除对象失败: %w", fileObject.Identity, err)
	}

	if err := lifecycle.FinalizeDeletion(impl.DBService.WithContext(operationCtx), fileObject.ID, lease.Token); err != nil {
		logger.Errorf("identity=%s stage=finalize", fileObject.Identity)
		return fmt.Errorf("文件 %s 完成删除失败: %w", fileObject.Identity, err)
	}
	return nil
}