feat: lots of stuff

5 files changed, 226 insertions, 4 deletions

diff --git a/src/marlin.agents.shell/CommandRunner.cs b/src/marlin.agents.shell/CommandRunner.cs
new file mode 100644
index 0000000..d4fcbb3
--- /dev/null
+++ b/src/marlin.agents.shell/CommandRunner.cs
@@ -0,0 +1,38 @@
+using System.Diagnostics;
+
+public static class CommandRunner
+{
+    public static async Task<int> RunAsync(string command, Action<string> onStdout, Action<string> onStderr)
+    {
+        var psi = new ProcessStartInfo
+        {
+            FileName = GetShell(),
+            RedirectStandardInput = true,
+            RedirectStandardOutput = true,
+            RedirectStandardError = true,
+            UseShellExecute = false,
+            CreateNoWindow = true
+        };
+
+        var process = new Process { StartInfo = psi };
+        process.OutputDataReceived += (_, e) => { if (e.Data != null) onStdout(e.Data); };
+        process.ErrorDataReceived  += (_, e) => { if (e.Data != null) onStderr(e.Data); };
+
+        process.Start();
+        process.BeginOutputReadLine();
+        process.BeginErrorReadLine();
+
+        // Write the command to stdin and close it
+        await process.StandardInput.WriteLineAsync(command);
+        await process.StandardInput.WriteLineAsync("exit $?");  // Exit with the command's exit code
+        process.StandardInput.Close();
+        
+        await process.WaitForExitAsync();
+        return process.ExitCode;
+    }
+
+    private static string GetShell()
+    {
+        return OperatingSystem.IsWindows() ? "cmd.exe" : "/bin/bash";
+    }
+}
\ No newline at end of file
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
diff --git a/src/marlin.agents.shell/MarlinApiClient.cs b/src/marlin.agents.shell/MarlinApiClient.cs
index aa4f013..900101c 100644
--- a/src/marlin.agents.shell/MarlinApiClient.cs
+++ b/src/marlin.agents.shell/MarlinApiClient.cs
@@ -1,16 +1,28 @@
 namespace marlin.agents.shell;
 
 using System.Net.Http.Json;
+using core.contracts;
 using core.models;
+using Grpc.Core;
+using Grpc.Net.Client;
+using ProtoBuf.Grpc.Client;
 
 public class MarlinApiClient
 {
+    private readonly string _baseUrl;
+    private readonly Guid _agentId;
+    private readonly string _agentToken;
+    
     private readonly HttpClient _http;
 
     public MarlinApiClient(string baseUrl, Guid agentId, string agentToken)
     {
-        _http = new HttpClient { BaseAddress = new Uri(baseUrl) };
-        _http.DefaultRequestHeaders.Add("Agent", $"{agentId} {agentToken}");
+        this._baseUrl = baseUrl;
+        this._agentId = agentId;
+        this._agentToken = agentToken;
+
+        _http = new HttpClient { BaseAddress = new Uri(this._baseUrl) };
+        _http.DefaultRequestHeaders.Add("Agent", $"{this._agentId} {this._agentToken}");
     }
     
     public static MarlinApiClient FromEnvironment()
@@ -35,4 +47,21 @@ public class MarlinApiClient
 
     public async Task<DispatchJob?> QueueNextJob(CancellationToken ct = default) 
         => await _http.GetFromJsonAsync<DispatchJob?>("api/dispatch/queue", ct);
+
+    public GrpcChannel CreateJobScopedRpc(DispatchJob job)
+    {
+        var credentials = CallCredentials.FromInterceptor(
+            async (_, metadata) =>
+            {
+                metadata.Add("Agent", $"{this._agentId} {this._agentToken}");
+                metadata.Add("Job", $"{job.Id} {job.Token}");
+            });
+
+        var channel = GrpcChannel.ForAddress(this._baseUrl, new GrpcChannelOptions
+        {
+            Credentials = ChannelCredentials.Create(new SslCredentials(), credentials)
+        });
+
+        return channel;
+    }
 }
diff --git a/src/marlin.agents.shell/Program.cs b/src/marlin.agents.shell/Program.cs
index 02a0bc9..75ea8ad 100644
--- a/src/marlin.agents.shell/Program.cs
+++ b/src/marlin.agents.shell/Program.cs
@@ -1,9 +1,20 @@
 using System.Runtime.InteropServices;
+using System.Threading.Channels;
+using Grpc.Net.Client;
 using marlin.agents.shell;
+using marlin.core.contracts;
+using marlin.core.functional;
 using marlin.core.models;
+using marlin.core.models.meta;
+using marlin.core.models.plan;
+using marlin.core.util;
+using ProtoBuf.Grpc.Client;
 
 DotNetEnv.Env.NoClobber().TraversePath().Load();
 
+var jobInstanceDir = Path.Combine(CommonPaths.AgentRoot, "instance");
+Directory.CreateDirectory(jobInstanceDir);
+
 var client = MarlinApiClient.FromEnvironment();
 
 var capabilities = new CapabilitySet();
@@ -16,5 +27,17 @@ AgentConfiguration config = await client.Handshake(status);
 Console.WriteLine($"IAM: {config.Name}");
 Console.WriteLine($"MCJ: {config.MaxConcurrentJobs}");
 
-var job = await client.QueueNextJob();
-Console.WriteLine(job == null ? "No pending jobs." : $"Claimed job {job.Id}: {job.Recipe}");
+while (true)
+{
+    var job = await client.QueueNextJob();
+    Console.WriteLine(job == null ? "No pending jobs." : $"Claimed job {job.Id}: {job.Recipe}");
+
+    if (job != null)
+    {
+        GitBinaryClient gitClient = new();
+        var uow = new JobUnitOfWork(job, client, gitClient);
+        await uow.Run();
+    }
+
+    await Task.Delay(TimeSpan.FromSeconds(5));
+}
\ No newline at end of file
diff --git a/src/marlin.agents.shell/marlin.agents.shell.csproj b/src/marlin.agents.shell/marlin.agents.shell.csproj
index ff86d63..0278fa0 100644
--- a/src/marlin.agents.shell/marlin.agents.shell.csproj
+++ b/src/marlin.agents.shell/marlin.agents.shell.csproj
@@ -9,6 +9,8 @@
 
     <ItemGroup>
       <PackageReference Include="DotNetEnv" Version="3.2.0" />
+      <PackageReference Include="Grpc.Net.Client" Version="2.84.0" />
+      <PackageReference Include="protobuf-net.Grpc" Version="1.3.14" />
     </ItemGroup>
 
     <ItemGroup>