path: root/src/marlin.web/Services/VersionControlSyncService.cs

namespace marlin.web.Services;

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
{
    private readonly ILogger<VersionControlSyncService> _logger;
    private readonly IServiceProvider _services;
    
    private ICollection<VersionControlSource> _sources;
    private Dictionary<Guid, DateTime> _lastPoll;

    public VersionControlSyncService(ILogger<VersionControlSyncService> logger, IServiceProvider services)
    {
        this._logger = logger;
        this._services = services;
        this._sources = new List<VersionControlSource>();
        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");
            }
        }
    }

    private async Task Tick(CancellationToken ct)
    {
        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<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)
    {
        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);
    }
    
    private async Task Sweep(ApplicationDbContext dbContext, CancellationToken ct)
    {
        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));
        
        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;
        }
    }
}