up

src/Scanner/StellaOps.Scanner.Worker/Processing/ScanJobProcessor.cs

@@ -7,21 +7,24 @@ using StellaOps.Scanner.Reachability;
 
 namespace StellaOps.Scanner.Worker.Processing;
 
-public sealed class ScanJobProcessor
-{
+public sealed class ScanJobProcessor
+{
     private readonly IReadOnlyDictionary<string, IScanStageExecutor> _executors;
     private readonly ScanProgressReporter _progressReporter;
     private readonly ILogger<ScanJobProcessor> _logger;
     private readonly IReachabilityUnionPublisherService _reachabilityPublisher;
+    private readonly Replay.ReplayBundleFetcher _replayBundleFetcher;
 
     public ScanJobProcessor(
         IEnumerable<IScanStageExecutor> executors,
         ScanProgressReporter progressReporter,
         IReachabilityUnionPublisherService reachabilityPublisher,
+        Replay.ReplayBundleFetcher replayBundleFetcher,
         ILogger<ScanJobProcessor> logger)
     {
         _progressReporter = progressReporter ?? throw new ArgumentNullException(nameof(progressReporter));
         _reachabilityPublisher = reachabilityPublisher ?? throw new ArgumentNullException(nameof(reachabilityPublisher));
+        _replayBundleFetcher = replayBundleFetcher ?? throw new ArgumentNullException(nameof(replayBundleFetcher));
         _logger = logger ?? throw new ArgumentNullException(nameof(logger));
 
         var map = new Dictionary<string, IScanStageExecutor>(StringComparer.OrdinalIgnoreCase);
@@ -52,18 +55,17 @@ public sealed class ScanJobProcessor
     public async ValueTask ExecuteAsync(ScanJobContext context, CancellationToken cancellationToken)
     {
         ArgumentNullException.ThrowIfNull(context);
-        // Placeholder: reachability publisher will be fed once lifter outputs are routed here.
-        _ = _reachabilityPublisher;
+            await EnsureReplayBundleFetchedAsync(context, cancellationToken).ConfigureAwait(false);
 
         foreach (var stage in ScanStageNames.Ordered)
         {
             cancellationToken.ThrowIfCancellationRequested();
 
             if (!_executors.TryGetValue(stage, out var executor))
-            {
-                continue;
-            }
-
+            {
+                continue;
+            }
+
             await _progressReporter.ExecuteStageAsync(
                 context,
                 stage,
@@ -71,4 +73,19 @@ public sealed class ScanJobProcessor
                 cancellationToken).ConfigureAwait(false);
         }
     }
+
+    private async Task EnsureReplayBundleFetchedAsync(ScanJobContext context, CancellationToken cancellationToken)
+    {
+        if (context.Analysis.TryGet<Replay.ReplaySealedBundleMetadata>(ScanAnalysisKeys.ReplaySealedBundleMetadata, out var sealedMetadata) && sealedMetadata is not null)
+        {
+            // Already fetched in this context
+            if (!string.IsNullOrWhiteSpace(context.ReplayBundlePath) && File.Exists(context.ReplayBundlePath))
+            {
+                return;
+            }
+
+            var path = await _replayBundleFetcher.FetchAsync(sealedMetadata, cancellationToken).ConfigureAwait(false);
+            context.ReplayBundlePath = path;
+        }
+    }
 }