feat: vcs and secret management support

1 files changed, 189 insertions, 12 deletions

diff --git a/src/marlin.web/Services/VersionControlSyncService.cs b/src/marlin.web/Services/VersionControlSyncService.cs
index ba40662..bf189b0 100644
--- a/src/marlin.web/Services/VersionControlSyncService.cs
+++ b/src/marlin.web/Services/VersionControlSyncService.cs
@@ -4,8 +4,10 @@ using System.Diagnostics;
 using System.Text;
 using core.functional;
 using core.models;
+using core.util;
 using Data;
 using Microsoft.EntityFrameworkCore;
+using Models;
 
 public class VersionControlSyncService : BackgroundService
 {
@@ -13,38 +15,213 @@ public class VersionControlSyncService : BackgroundService
     private readonly IServiceProvider _services;
     
     private ICollection<VersionControlSource> _sources;
-    private Dictionary<Guid, DateTime> _nextPoll;
+    private Dictionary<Guid, DateTime> _lastPoll;
 
     public VersionControlSyncService(ILogger<VersionControlSyncService> logger, IServiceProvider services)
     {
         this._logger = logger;
         this._services = services;
         this._sources = new List<VersionControlSource>();
-        this._nextPoll = new();
+        this._lastPoll = new();
+    }
+    
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        await this.FetchSources(stoppingToken);
+
+        while (!stoppingToken.IsCancellationRequested)
+        { 
+            await Task.Delay(TimeSpan.FromSeconds(10), stoppingToken);
+
+            try
+            {
+                await Tick(stoppingToken);
+            }
+            catch (Exception ex) when (ex is not OperationCanceledException)
+            {
+                this._logger.LogError(ex, "VCS sync tick failed");
+            }
+        }
     }
 
-    public async Task FetchSources()
+    private async Task Tick(CancellationToken ct)
     {
-        await using var scoped = this._services.CreateAsyncScope();
-        var dbContext = scoped.ServiceProvider.GetRequiredService<ApplicationDbContext>();
-        this._sources = await dbContext.Sources.ToListAsync();
+        var now = DateTime.UtcNow;
+        var due = this
+            ._sources
+            .Where(x => this._lastPoll.TryGetValue(x.Id, out var lastTime) ? now >= lastTime.AddSeconds(x.MinPollingInterval) : true)
+            .ToArray();
+
+        if (due.Length > 0)
+        {
+            await using var scoped = this._services.CreateAsyncScope();
+            var dbContext = scoped.ServiceProvider.GetRequiredService<ApplicationDbContext>();
+            var sourceSecretStore = scoped.ServiceProvider.GetRequiredService<ISourceSecretStore>();
+
+            foreach (var source in due)
+            {
+                var trackedSource = await dbContext.Sources
+                    .Include(x => x.Branches)
+                    .ThenInclude(x => x.Commit)
+                    .FirstAsync(x => x.Id == source.Id, ct);
+
+                try
+                {
+                    await this.SyncRefs(dbContext, sourceSecretStore, trackedSource, ct);
+                    await this.SyncBlobs(dbContext, sourceSecretStore, trackedSource, ct);
+                }
+                catch (Exception ex) when (ex is not OperationCanceledException)
+                {
+                    this._logger.LogError(ex, "Failed to sync vcs source '{name}'", source.UpstreamUrl);
+                }
+                finally
+                {
+                    this._lastPoll[source.Id] = DateTime.UtcNow;
+                }
+            }
+
+            if (dbContext.ChangeTracker.HasChanges())
+            {
+                await this.Sweep(dbContext, ct);
+                
+                await dbContext.SaveChangesAsync(ct);
+                this._logger.LogInformation("Synced {count} vcs roots", due.Length);   
+            }
+        }
     }
 
-    private async Task SyncSource(VersionControlSource source, ApplicationDbContext db)
+    private async Task<IGitClient> GetClient(VersionControlSource source, ISourceSecretStore store, CancellationToken ct)
+    {
+        IGitCredential? cred = null;
+        if (source.AccessSecretId != null)
+        {
+            ISecret? secret = await store.GetAsync(source.AccessSecretId.Value, ct);
+            if (secret == null)
+                throw new InvalidOperationException($"Could not find access secret for '{source.UpstreamUrl}'");
+
+            if (secret.Kind == SecretKind.SSHKey && secret.Has("public_key", "private_key"))
+                cred = new SshKeyGitCredential(secret.Content["private_key"]);
+            
+            else if (secret.Kind == SecretKind.HTTPToken && secret.Has("identity", "token"))
+                cred = new TokenGitCredential(secret.Content["token"], secret.Content["identity"]);
+            
+            else
+                throw new InvalidOperationException($"Could not resolve access secret content to credentials for '{source.UpstreamUrl}'");
+        }
+        
+        IGitClient client = new GitBinaryClient(cred);
+        return client;
+    }
+
+    private async Task<string> GetLocalPath(IGitClient client, VersionControlSource source)
+    {
+        var dir = Path.Join(Environment.CurrentDirectory, "run", "sources", source.Id.ToString("N"));
+        if (!Directory.Exists(dir))
+        {
+            Directory.CreateDirectory(dir);
+            await client.Init(dir);
+        }
+
+        return dir;
+    }
+    
+    public async Task FetchSources(CancellationToken ct)
     {
-        IGitClient client = new GitBinaryClient();
-        var refs = await client.GetRemoteRefs(source.UpstreamUrl);
+        await using var scoped = this._services.CreateAsyncScope();
+        var dbContext = scoped.ServiceProvider.GetRequiredService<ApplicationDbContext>();
         
+        this._sources = await dbContext.Sources
+            .Include(x => x.Branches)
+            .ToListAsync(ct);
         
+        await this.Sweep(dbContext, ct);
     }
     
-    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    private async Task Sweep(ApplicationDbContext dbContext, CancellationToken ct)
     {
-        await this.FetchSources();
+        await dbContext.SourceCommits
+            .Include(x => x.Branches)
+            .Where(x => !x.Branches.Any())
+            .ExecuteDeleteAsync(ct);
+    }
+
+    private async Task SyncRefs(ApplicationDbContext dbContext, ISourceSecretStore secretStore, VersionControlSource source, CancellationToken ct)
+    {
+        IGitClient client = await this.GetClient(source, secretStore, ct);
+        var newRefs = await client.GetRemoteRefs(source.UpstreamUrl);
+
+        Dictionary<string, string> currentRefs = source.Branches
+            .ToDictionary(x => x.Name, x => x.Commit.Hash);
+        Dictionary<string, VersionControlledBranch> branchesByName = source.Branches
+            .ToDictionary(x => x.Name);
+
+        var added = newRefs.Keys.Except(currentRefs.Keys).ToArray();
+        var removed = currentRefs.Keys.Except(newRefs.Keys).ToArray();
+        var changed = currentRefs.Keys.Intersect(newRefs.Keys)
+            .Where(name => currentRefs[name] != newRefs[name])
+            .ToArray();
+
+        foreach (var name in removed)
+            source.Branches.Remove(branchesByName[name]);
+
+        // Pre-load the commits these refs now point to, so a hash shared across
+        // multiple branches (or seen in an earlier sync) resolves to one row
+        // instead of racing duplicate inserts.
+        var neededHashes = added.Concat(changed)
+            .Select(name => newRefs[name])
+            .Distinct()
+            .ToHashSet();
+
+        var commitsByHash = await dbContext.SourceCommits
+            .Where(c => c.SourceId == source.Id && neededHashes.Contains(c.Hash))
+            .ToDictionaryAsync(c => c.Hash, ct);
+
+        VersionControlledCommit GetOrCreateCommit(string hash)
+        {
+            if (!commitsByHash.TryGetValue(hash, out var commit))
+            {
+                commit = new VersionControlledCommit { SourceId = source.Id, Hash = hash, BlobsFetched = false };
+                commitsByHash[hash] = commit;
+            }
+
+            return commit;
+        }
+
+        foreach (var name in added)
+            source.Branches.Add(new VersionControlledBranch { Name = name, Commit = GetOrCreateCommit(newRefs[name]) });
+
+        foreach (var name in changed)
+            branchesByName[name].Commit = GetOrCreateCommit(newRefs[name]);
+    }
+    
+    private async Task SyncBlobs(ApplicationDbContext dbContext, ISourceSecretStore secretStore, VersionControlSource source, CancellationToken ct)
+    {
+        var refs = source.Branches
+            .Where(x => !x.Commit.BlobsFetched)
+            .DistinctBy(x => x.CommitId)
+            .ToArray();
+
+        IGitClient client = await this.GetClient(source, secretStore, ct);
+        var localPath = await GetLocalPath(client, source);
+        await client.FetchTrees(localPath, source.UpstreamUrl, refs.Select(x => x.Name));
         
-        while (!stoppingToken.IsCancellationRequested)
+        foreach (VersionControlledCommit commit in refs.Select(x => x.Commit))
         {
+            var blobNames = await client.ListTree(localPath, commit.Hash, ".");
+            
+            foreach (string blobName in blobNames)
+            {
+                var content = await client.CatFile(localPath, commit.Hash, blobName);
+                dbContext.SourceBlobs.Add(new VersionControlledBlob
+                {
+                    Source = source,
+                    Commit = commit,
+                    Path = blobName,
+                    Content = content
+                });
+            }
             
+            commit.BlobsFetched = true;
         }
     }
 }
\ No newline at end of file