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 _logger; public StorageMigrationWorker(IServiceScopeFactory scopeFactory, ILogger 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(); await db.Updateable() .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().Where(x => x.TargetStorageType == MediaStorageType.OpenList && x.Status == StorageMigrationTaskStatus.Running).ToListAsync(); foreach (var task in running) { await db.Updateable() .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() .SetColumns(x => new StorageMigrationTask { Status = StorageMigrationTaskStatus.Queued, CurrentFile = null, CurrentVideoId = null, UpdatedAt = DateTime.Now }) .Where(x => x.Id == task.Id).ExecuteCommandAsync(); } await db.Updateable() .SetColumns(x => new StorageMigrationTask { Status = StorageMigrationTaskStatus.PartiallyFailed, ErrorMessage = "旧文件清理被应用重启中断,请确认后重试清理。", UpdatedAt = DateTime.Now }) .Where(x => x.Status == StorageMigrationTaskStatus.Cleaning) .ExecuteCommandAsync(); } private async Task RunOneBatchAsync(CancellationToken cancellationToken) { string taskId; int concurrency; List itemIds; using (var scope = _scopeFactory.CreateScope()) { var db = scope.ServiceProvider.GetRequiredService(); var task = await db.Queryable() .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() .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().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().RefreshCountsAsync(taskId); return true; } private async Task ProcessInScopeAsync(string taskId, string itemId, CancellationToken cancellationToken) { using var scope = _scopeFactory.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); var item = await db.Queryable().InSingleAsync(itemId); var videoId = item == null ? null : item.VideoId; var oldVideoPath = item == null ? null : item.OldVideoPath; var now = DateTime.Now; await db.Updateable() .SetColumns(x => new StorageMigrationTask { CurrentVideoId = videoId, CurrentFile = oldVideoPath, UpdatedAt = now }) .Where(x => x.Id == taskId && x.Status == StorageMigrationTaskStatus.Running).ExecuteCommandAsync(); await scope.ServiceProvider.GetRequiredService().ProcessAsync(itemId, cancellationToken); } private async Task CompleteTaskAsync(string taskId) { using var scope = _scopeFactory.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); var task = await db.Queryable().InSingleAsync(taskId); if (task?.Status != StorageMigrationTaskStatus.Running) return; var failed = await db.Queryable().Where(x => x.TaskId == taskId && x.Stage == StorageMigrationItemStage.Failed).CountAsync(); await db.Updateable() .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().RefreshCountsAsync(taskId); } } }