douyin/service/StorageMigrationWorker.cs

using dy.net.model.dto;
using dy.net.model.entity;
using Microsoft.Extensions.DependencyInjection;
using SqlSugar;
using MediaStorageType = dy.net.model.dto.StorageType;

namespace dy.net.service
{
    public class StorageMigrationWorker : BackgroundService
    {
        private readonly IServiceScopeFactory _scopeFactory;
        private readonly ILogger<StorageMigrationWorker> _logger;

        public StorageMigrationWorker(IServiceScopeFactory scopeFactory, ILogger<StorageMigrationWorker> logger)
        {
            _scopeFactory = scopeFactory;
            _logger = logger;
        }

        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            await RecoverInterruptedAsync();
            while (!stoppingToken.IsCancellationRequested)
            {
                try
                {
                    var worked = await RunOneBatchAsync(stoppingToken);
                    if (!worked) await Task.Delay(TimeSpan.FromSeconds(2), stoppingToken);
                }
                catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { }
                catch (Exception ex)
                {
                    _logger.LogError(ex, "存储迁移后台服务发生错误");
                    await Task.Delay(TimeSpan.FromSeconds(3), stoppingToken);
                }
            }
        }

        private async Task RecoverInterruptedAsync()
        {
            using var scope = _scopeFactory.CreateScope();
            var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>();
            await db.Updateable<StorageMigrationTask>()
                .SetColumns(x => new StorageMigrationTask
                {
                    Status = StorageMigrationTaskStatus.Cancelled,
                    ErrorMessage = "旧 WebDAV 迁移任务已停止，仅保留审计信息；请创建新的 OpenList 接管任务。",
                    CurrentFile = null,
                    CurrentVideoId = null,
                    CompletedAt = DateTime.Now,
                    UpdatedAt = DateTime.Now
                })
                .Where(x => x.TargetStorageType == null
                    && (x.Status == StorageMigrationTaskStatus.Queued
                        || x.Status == StorageMigrationTaskStatus.Running
                        || x.Status == StorageMigrationTaskStatus.Paused
                        || x.Status == StorageMigrationTaskStatus.Cleaning))
                .ExecuteCommandAsync();

            var running = await db.Queryable<StorageMigrationTask>().Where(x =>
                x.TargetStorageType == MediaStorageType.OpenList
                && x.Status == StorageMigrationTaskStatus.Running).ToListAsync();
            foreach (var task in running)
            {
                await db.Updateable<StorageMigrationItem>()
                    .SetColumns(x => new StorageMigrationItem { Stage = StorageMigrationItemStage.Pending, UpdatedAt = DateTime.Now })
                    .Where(x => x.TaskId == task.Id && (x.Stage == StorageMigrationItemStage.Uploading
                        || x.Stage == StorageMigrationItemStage.Verifying || x.Stage == StorageMigrationItemStage.Committing))
                    .ExecuteCommandAsync();
                await db.Updateable<StorageMigrationTask>()
                    .SetColumns(x => new StorageMigrationTask { Status = StorageMigrationTaskStatus.Queued, CurrentFile = null, CurrentVideoId = null, UpdatedAt = DateTime.Now })
                    .Where(x => x.Id == task.Id).ExecuteCommandAsync();
            }
            await db.Updateable<StorageMigrationTask>()
                .SetColumns(x => new StorageMigrationTask
                {
                    Status = StorageMigrationTaskStatus.PartiallyFailed,
                    ErrorMessage = "旧文件清理被应用重启中断，请确认后重试清理。",
                    UpdatedAt = DateTime.Now
                })
                .Where(x => x.Status == StorageMigrationTaskStatus.Cleaning)
                .ExecuteCommandAsync();
        }

        private async Task<bool> RunOneBatchAsync(CancellationToken cancellationToken)
        {
            string taskId;
            int concurrency;
            List<string> itemIds;
            using (var scope = _scopeFactory.CreateScope())
            {
                var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>();
                var task = await db.Queryable<StorageMigrationTask>()
                    .Where(x => x.TargetStorageType == MediaStorageType.OpenList
                        && (x.Status == StorageMigrationTaskStatus.Queued || x.Status == StorageMigrationTaskStatus.Running))
                    .OrderBy(x => x.CreatedAt).FirstAsync();
                if (task == null) return false;
                taskId = task.Id;
                concurrency = Math.Clamp(task.Concurrency, 1, 3);
                if (task.Status == StorageMigrationTaskStatus.Queued)
                {
                    await db.Updateable<StorageMigrationTask>()
                        .SetColumns(x => new StorageMigrationTask { Status = StorageMigrationTaskStatus.Running, StartedAt = DateTime.Now, UpdatedAt = DateTime.Now })
                        .Where(x => x.Id == task.Id && x.Status == StorageMigrationTaskStatus.Queued).ExecuteCommandAsync();
                }
                itemIds = await db.Queryable<StorageMigrationItem>().Where(x => x.TaskId == task.Id && x.Stage == StorageMigrationItemStage.Pending)
                    .OrderBy(x => x.CreatedAt).Take(concurrency).Select(x => x.Id).ToListAsync();
            }

            if (itemIds.Count == 0)
            {
                await CompleteTaskAsync(taskId);
                return true;
            }

            await Task.WhenAll(itemIds.Select(itemId => ProcessInScopeAsync(taskId, itemId, cancellationToken)));
            using (var scope = _scopeFactory.CreateScope())
                await scope.ServiceProvider.GetRequiredService<StorageMigrationService>().RefreshCountsAsync(taskId);
            return true;
        }

        private async Task ProcessInScopeAsync(string taskId, string itemId, CancellationToken cancellationToken)
        {
            using var scope = _scopeFactory.CreateScope();
            var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>();
            var item = await db.Queryable<StorageMigrationItem>().InSingleAsync(itemId);
            var videoId = item == null ? null : item.VideoId;
            var oldVideoPath = item == null ? null : item.OldVideoPath;
            var now = DateTime.Now;
            await db.Updateable<StorageMigrationTask>()
                .SetColumns(x => new StorageMigrationTask { CurrentVideoId = videoId, CurrentFile = oldVideoPath, UpdatedAt = now })
                .Where(x => x.Id == taskId && x.Status == StorageMigrationTaskStatus.Running).ExecuteCommandAsync();
            await scope.ServiceProvider.GetRequiredService<StorageMigrationItemProcessor>().ProcessAsync(itemId, cancellationToken);
        }

        private async Task CompleteTaskAsync(string taskId)
        {
            using var scope = _scopeFactory.CreateScope();
            var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>();
            var task = await db.Queryable<StorageMigrationTask>().InSingleAsync(taskId);
            if (task?.Status != StorageMigrationTaskStatus.Running) return;
            var failed = await db.Queryable<StorageMigrationItem>().Where(x => x.TaskId == taskId && x.Stage == StorageMigrationItemStage.Failed).CountAsync();
            await db.Updateable<StorageMigrationTask>()
                .SetColumns(x => new StorageMigrationTask
                {
                    Status = failed == 0 ? StorageMigrationTaskStatus.Completed : StorageMigrationTaskStatus.PartiallyFailed,
                    CurrentVideoId = null,
                    CurrentFile = null,
                    CompletedAt = DateTime.Now,
                    UpdatedAt = DateTime.Now
                }).Where(x => x.Id == taskId && x.Status == StorageMigrationTaskStatus.Running).ExecuteCommandAsync();
            await scope.ServiceProvider.GetRequiredService<StorageMigrationService>().RefreshCountsAsync(taskId);
        }
    }
}