diff options
Diffstat (limited to 'src/marlin.web/Services/VersionControlSyncService.cs')
| -rw-r--r-- | src/marlin.web/Services/VersionControlSyncService.cs | 201 |
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 |