refactor: dispatch import jobs through MediatR

src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs

@@ -1,48 +1,37 @@
+using MediatR;
 using Microsoft.Extensions.Options;
-using Workflow.Application.BuildFlow.Imports.Services;
+using Workflow.Application.BuildFlow.Imports.Commands;
-using Workflow.Application.Notifications;
+using Workflow.Application.Reliability;
-using Workflow.Domain.Enums;
-using Workflow.Infrastructure.Persistence;
 
 namespace BuildFlow.Host.Services.BackgroundServices;
 
 /// <summary>
-/// 导入任务后台处理器：周期性扫描 Pending 任务，置为 Running 后调用 ImportJobProcessor。
+/// 周期触发导入任务原子领取命令的 Host 后台服务。
-/// 参考 TimeoutSchedulerService 结构：BackgroundService 单例 + IServiceScopeFactory 创建独立 scope，
-/// 以便解析 scoped 依赖（WorkflowDbContext、IImportJobProcessor）。
 /// </summary>
-public class ImportJobProcessorService : BackgroundService
+public sealed class ImportJobProcessorService(
-{
-    private readonly IServiceScopeFactory _scopeFactory;
-    private readonly ILogger<ImportJobProcessorService> _logger;
-    private readonly IOptionsMonitor<NotificationOptions> _options;
-
-    public ImportJobProcessorService(
     IServiceScopeFactory scopeFactory,
-        ILogger<ImportJobProcessorService> logger,
+    IOptionsMonitor<WorkflowJobOptions> options,
-        IOptionsMonitor<NotificationOptions> options)
+    ILogger<ImportJobProcessorService> logger) : BackgroundService
-    {
+{
-        _scopeFactory = scopeFactory;
+    private readonly string _leaseOwner =
-        _logger = logger;
+        $"import:{Environment.MachineName}:{Environment.ProcessId}:{Guid.NewGuid():N}";
-        _options = options;
-    }
 
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
-        // timer 间隔启动时定一次（方案 A：timer 参数不热重载，改间隔需重启服务）
+        var importOptions = options.CurrentValue.Import;
-        var opts = _options.CurrentValue;
+        logger.LogInformation(
-        _logger.LogInformation("ImportJobProcessor 启动，轮询间隔 {Interval}s", opts.Scheduler.PollIntervalSeconds);
+            "ImportJobProcessor 启动，Owner={LeaseOwner}，轮询间隔 {Interval}s",
+            _leaseOwner,
+            importOptions.PollIntervalSeconds);
 
-        // 启动时先等待一轮，避免与应用启动并发争抢资源
-        await Task.Delay(opts.Scheduler.StartupDelaySeconds * 1000, stoppingToken);
-
-        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(opts.Scheduler.PollIntervalSeconds));
-
-        // 整个 do-while 纳入 try/catch：timer.WaitForNextTickAsync 在 stoppingToken 取消时
-        // 会抛 OperationCanceledException，该异常发生在 while 条件求值处（不在内层 try 范围），
-        // 若不捕获会逃逸出 ExecuteAsync，触发 BackgroundServiceExceptionBehavior=StopHost 导致进程崩溃。
         try
         {
+            await Task.Delay(
+                TimeSpan.FromSeconds(importOptions.StartupDelaySeconds),
+                stoppingToken);
+            using var timer = new PeriodicTimer(
+                TimeSpan.FromSeconds(importOptions.PollIntervalSeconds));
+
             do
             {
                 try
@@ -55,45 +44,41 @@ public class ImportJobProcessorService : BackgroundService
                 }
                 catch (Exception ex)
                 {
-                    _logger.LogError(ex, "ImportJobProcessor 单轮处理异常");
+                    logger.LogError(ex, "ImportJobProcessor 单轮处理异常");
                 }
             }
             while (await timer.WaitForNextTickAsync(stoppingToken));
         }
         catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
         {
-            // 正常停机：WaitForNextTickAsync 被取消，优雅退出
+            // 正常停机。
         }
 
-        _logger.LogInformation("ImportJobProcessor 停止");
+        logger.LogInformation("ImportJobProcessor 停止，Owner={LeaseOwner}", _leaseOwner);
     }
 
-    /// <summary>执行一轮扫描：拉取 Pending 任务 ID 列表，逐个调用 IImportJobProcessor.ProcessAsync。</summary>
+    /// <summary>执行一轮导入任务领取和处理。</summary>
-    public async Task RunScanAsync(CancellationToken ct = default)
+    public async Task<int> RunScanAsync(CancellationToken ct = default)
     {
-        using var scope = _scopeFactory.CreateScope();
+        using var scope = scopeFactory.CreateScope();
-        var db = scope.ServiceProvider.GetRequiredService<WorkflowDbContext>();
+        var mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
-        var processor = scope.ServiceProvider.GetRequiredService<IImportJobProcessor>();
+        var importOptions = options.CurrentValue.Import;
+        var processed = await mediator.Send(
+            new ProcessAvailableImportJobsCommand(
+                _leaseOwner,
+                importOptions.BatchSize,
+                TimeSpan.FromSeconds(importOptions.LeaseSeconds),
+                importOptions.MaxAttempts),
+            ct);
 
-        var pendingIds = await db.BuildFlowImportJobs
+        if (processed > 0)
-            .AsNoTracking()
+        {
-            .Where(j => j.Status == BuildFlowImportJobStatus.Pending)
+            logger.LogInformation(
-            .OrderBy(j => j.CreatedAt)
+                "ImportJobProcessor 完成一轮处理，Owner={LeaseOwner}，领取 {Count} 个任务",
-            .Select(j => j.Id)
+                _leaseOwner,
-            .Take(20)
+                processed);
-            .ToListAsync(ct);
+        }
 
-        foreach (var jobId in pendingIds)
+        return processed;
-        {
-            ct.ThrowIfCancellationRequested();
-            try
-            {
-                await processor.ProcessAsync(jobId, ct);
-            }
-            catch (Exception ex)
-            {
-                _logger.LogError(ex, "导入任务 {JobId} 处理失败", jobId);
-            }
-        }
     }
 }

src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs

@@ -9,8 +9,4 @@ public interface IImportJobProcessor
         string leaseOwner,
         TimeSpan leaseDuration,
         CancellationToken ct);
-
-    /// <summary>旧 Host 扫描器的临时兼容入口，Host 改为 MediatR 后删除。</summary>
-    Task ProcessAsync(Guid jobId, CancellationToken ct)
-        => ProcessAsync(jobId, "legacy-import-host", TimeSpan.FromMinutes(5), ct);
 }

tests/BuildFlow.Architecture.Tests/WorkflowJobArchitectureTests.cs

@@ -29,6 +29,21 @@ public class WorkflowJobArchitectureTests
         Assert.DoesNotContain("ProcessEngine", source);
     }
 
+    [Fact]
+    public void ImportJobProcessorServiceShouldUseMediatorWithoutWorkflowInfrastructure()
+    {
+        var source = ReadSource(
+            "src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs");
+
+        Assert.Contains("IMediator", source);
+        Assert.Contains("mediator.Send", source);
+        Assert.Contains("ProcessAvailableImportJobsCommand", source);
+        Assert.DoesNotContain("WorkflowDbContext", source);
+        Assert.DoesNotContain("IImportJobProcessor", source);
+        Assert.DoesNotContain("BuildFlowImportJobStatus", source);
+        Assert.DoesNotContain(".Where(", source);
+    }
+
     private static string ReadSource(string relativePath)
     {
         var path = Path.Combine(