1 files changed, 130 insertions, 0 deletions
diff --git a/src/marlin.agents.shell/JobUnitOfWork.cs b/src/marlin.agents.shell/JobUnitOfWork.cs
new file mode 100644
index 0000000..85df806
--- /dev/null
+++ b/src/marlin.agents.shell/JobUnitOfWork.cs
@@ -0,0 +1,130 @@
+namespace marlin.agents.shell;
+
+using System.Threading.Channels;
+using core.contracts;
+using core.functional;
+using core.models;
+using core.models.meta;
+using core.models.plan;
+using core.util;
+using Grpc.Net.Client;
+using ProtoBuf.Grpc.Client;
+
+public class JobUnitOfWork
+{
+    private static string _jobInstanceDir = Path.Combine(CommonPaths.AgentRoot, "instance");
+    
+    private readonly MarlinApiClient _apiClient;
+    private readonly IGitClient _gitClient;
+    private readonly DispatchJob _job;
+    private readonly string _checkoutPath;
+
+    private GrpcChannel _channel;
+    private IInstanceService _instanceService;
+    
+    private Channel<LogEventRequest> _logQueue;
+
+    public JobUnitOfWork(DispatchJob job, MarlinApiClient apiClient, IGitClient gitClient)
+    {
+        this._job = job;
+        this._apiClient = apiClient;
+        this._gitClient = gitClient;
+        
+        this._checkoutPath = Path.Combine(_jobInstanceDir, job.Id.ToString("N"));
+    }
+
+    public async Task Run()
+    {
+        this._channel = this._apiClient.CreateJobScopedRpc(this._job);
+        this._instanceService = this._channel.CreateGrpcService<IInstanceService>();
+        this._logQueue = Channel.CreateUnbounded<LogEventRequest>();
+
+        var drainTask = Task.Run(this.DrainLogEvents);
+
+        await StatusUpdate(JobStatus.Active);
+        
+        await this.CheckoutSource();
+        var plan = await this.BuildPlan();
+        var success = await this.ExecutePlan(plan);
+
+        await StatusUpdate(success ? JobStatus.Success : JobStatus.Failed);
+
+        this._logQueue.Writer.Complete();
+        await drainTask;
+    }
+    
+    private async Task CheckoutSource()
+    {
+        Directory.CreateDirectory(_jobInstanceDir);
+        if (Path.Exists(this._checkoutPath))
+            Directory.Delete(this._checkoutPath, true);
+        
+        await this._gitClient.Checkout(this._checkoutPath, this._job.Vcs, this._job.Ref, this._job.Commit);
+    }
+
+    private async Task DrainLogEvents()
+    {
+        await foreach (var evt in this._logQueue.Reader.ReadAllAsync())
+            await this._instanceService.SendLogEvent(evt);
+    }
+
+    private async Task<ExecutionPlan> BuildPlan()
+    {
+        WorkflowBuilder builder = new();
+        builder.LoadFromDirectory(Path.Join(this._checkoutPath, ".marlin"));
+
+        Workflow workflow = builder.Build();
+        Recipe recipe = workflow.Recipes[this._job.Recipe];
+        ExecutionPlan plan = ExecutionPlan.Create(workflow, recipe);
+
+        return plan;
+    }
+
+    private async Task<bool> ExecutePlan(ExecutionPlan plan)
+    {
+        Dictionary<string, CommandResult> results = new();
+    
+        foreach (var component in plan.Components)
+        {
+            if (component.ResultSelector != null)
+            {
+                if (!results.ContainsKey(component.ResultSelector.Command.Id))
+                    // prerequisite step did not run
+                    continue;
+
+                var storedResult = results[component.ResultSelector.Command.Id];
+                if (!storedResult.Matches(component.ResultSelector.Result))
+                    // prerequisite step returned incorrect error code
+                    continue;
+            }
+
+            Console.WriteLine(component);
+            Task<int> run = CommandRunner.RunAsync(
+                component.Command,
+                o => Log(o, false, component),
+                o => Log(o, true, component));
+            
+            var result = await run;
+            results[component.Id] = result == 0 ? CommandResult.Success : CommandResult.Failure;
+        }
+
+        return results.All(x => x.Value == CommandResult.Success);
+    }
+    
+    private void Log(string message, bool error, ExecutionCommand component)
+    {
+        this._logQueue.Writer.TryWrite(
+            new LogEventRequest
+            {
+                At = DateTime.UtcNow,
+                Error = error,
+                Message = message,
+                StepId = component.Id
+            });
+    }
+
+    private async Task StatusUpdate(JobStatus status)
+    {
+        await this._instanceService.SendStatusUpdate(new StatusUpdateRequest { Status = status });
+    }
+}
+\ No newline at end of file