w4c-workflows-api/Services/WorkflowSyncService.cs

using Microsoft.EntityFrameworkCore;
using w4c_workflows.Data;
using w4c_workflows.Models;

namespace w4c_workflows.Services;

public sealed record SyncError(string Path, IReadOnlyList<string> Errors);

public sealed class SyncResult
{
    public string? HeadSha { get; set; }
    public bool Dirty { get; set; }
    public int Compiled { get; set; }
    public int Removed { get; set; }
    public List<SyncError> Errors { get; } = new();
}

/// <summary>
/// Re-reads <c>workflows/*.yaml</c> from the git source, compiles each to the
/// tenant's DB state, and removes workflows whose files were deleted. Ids are
/// deterministic, so re-syncing upserts instead of duplicating. Files that fail
/// to compile are reported in <see cref="SyncResult.Errors"/> and not persisted.
///
/// When Forgejo-backed mode is active, the service pulls the latest changes from
/// the remote before reading files, ensuring the compiled state reflects the
/// latest committed YAML.
/// </summary>
public class WorkflowSyncService
{
    private readonly WorkflowsDbContext _db;
    private readonly WorkflowCompiler _compiler;
    private readonly IWorkflowSource _source;
    private readonly WorkflowSourceFactory? _sourceFactory;
    private readonly ILogger<WorkflowSyncService> _logger;
    private readonly SyncGate? _gate;
    private readonly ILeaseService? _leases;

    public WorkflowSyncService(
        WorkflowsDbContext db,
        WorkflowCompiler compiler,
        IWorkflowSource source,
        ILogger<WorkflowSyncService> logger,
        WorkflowSourceFactory? sourceFactory = null,
        SyncGate? gate = null,
        ILeaseService? leases = null)
    {
        _db = db;
        _compiler = compiler;
        _source = source;
        _logger = logger;
        _sourceFactory = sourceFactory;
        _gate = gate;
        _leases = leases;
    }

    /// <summary>
    /// Serializes the sync against other syncs of the same <c>(tenant, repo)</c>:
    /// an in-process gate for this replica plus a best-effort distributed lease for
    /// other replicas. Without this, two overlapping syncs insert the same
    /// deterministic ids and collide on the primary key (P1-9).
    /// </summary>
    public async Task<SyncResult> SyncAsync(string tenantId, string repoName, CancellationToken ct)
    {
        using var gate = _gate == null ? null : await _gate.AcquireAsync(tenantId, repoName, ct);

        var owner = Guid.NewGuid().ToString("N");
        const string leaseKey = "sync";
        var held = false;
        if (_leases != null)
        {
            // Wait up to ~2s for another replica to finish, then proceed best-effort:
            // the deterministic ids make a subsequent sync an upsert, not a duplicate.
            for (var attempt = 0; attempt < 20 && !held; attempt++)
            {
                held = await _leases.AcquireAsync(tenantId, $"{leaseKey}:{repoName}", owner, TimeSpan.FromMinutes(2));
                if (!held)
                    await Task.Delay(100, ct);
            }

            if (!held)
                _logger.LogDebug("Sync lease for tenant {TenantId} repo {Repo} was busy; proceeding best-effort", tenantId, repoName);
        }

        try
        {
            return await SyncCoreAsync(tenantId, repoName, ct);
        }
        finally
        {
            if (held)
                await _leases!.ReleaseAsync(tenantId, $"{leaseKey}:{repoName}", owner);
        }
    }

    private async Task<SyncResult> SyncCoreAsync(string tenantId, string repoName, CancellationToken ct)
    {
        // Resolve the source for the SELECTED repo. The request-scoped `_source`
        // was built from whatever repo was current when the request started, which
        // is the wrong repo immediately after a repository switch. In Forgejo-backed
        // mode, creating the source ensures the per-user clone exists and is up to
        // date (clone + pull), so a switch always reads the newly selected repo.
        var source = _source;
        if (_sourceFactory?.IsForgejoBacked == true)
        {
            _logger.LogDebug("Resolving workflow source for tenant {TenantId}, repo {Repo}", tenantId, repoName);
            source = await _sourceFactory.CreateAsync(tenantId, repoName, ct);
        }

        var state = await source.GetStateAsync(ct);
        var files = await source.ListAsync(ct);

        var result = new SyncResult { HeadSha = state.HeadSha, Dirty = state.Dirty };

        // Only the CURRENT repo's workflows are compiled/kept active. Workflows from
        // a previously-selected repo stay in the DB but are excluded from active
        // listings (their <c>Repo</c> differs), so switching repos unloads them
        // without destroying their history.
        var existing = await _db.Workflows
            .Include(w => w.Tasks)
            .Include(w => w.TaskEdges)
            .Where(w => w.TenantId == tenantId && w.Repo == repoName)
            .ToListAsync(ct);
        var existingByPath = existing.ToDictionary(w => w.Path, StringComparer.Ordinal);

        var seenPaths = new HashSet<string>(StringComparer.Ordinal);
        var seenNames = new HashSet<string>(StringComparer.Ordinal);

        foreach (var file in files)
        {
            ct.ThrowIfCancellationRequested();
            seenPaths.Add(file.Path);

            string content;
            try
            {
                content = await source.ReadAsync(file.Path, ct);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Failed to read workflow file {Path}", file.Path);
                result.Errors.Add(new SyncError(file.Path, new[] { $"read error: {ex.Message}" }));
                continue;
            }

            var compiled = _compiler.Compile(content, file.Path, tenantId, repoName);
            if (!compiled.Success)
            {
                result.Errors.Add(new SyncError(file.Path, compiled.Errors));
                continue;
            }

            var name = compiled.Workflow!.Workflow.Name;
            if (!seenNames.Add(name))
            {
                result.Errors.Add(new SyncError(file.Path, new[] { $"duplicate workflow name '{name}'" }));
                continue;
            }

            var existingWorkflow = existingByPath.GetValueOrDefault(file.Path);
            await UpsertAsync(existingWorkflow, compiled.Workflow!, state.HeadSha, ct);
            result.Compiled++;
        }

        // Remove workflows whose files no longer exist in git.
        foreach (var workflow in existing)
        {
            if (seenPaths.Contains(workflow.Path)) continue;
            await RemoveWorkflowAsync(workflow, ct);
            result.Removed++;
            _logger.LogInformation("Removed workflow {Name} ({Path}) for tenant {TenantId} — file deleted",
                workflow.Name, workflow.Path, tenantId);
        }

        await _db.SaveChangesAsync(ct);
        return result;
    }

    private async Task UpsertAsync(Workflow? existing, CompiledWorkflow compiled, string? gitSha, CancellationToken ct)
    {
        var wf = compiled.Workflow;
        var now = DateTime.UtcNow;

        if (existing == null)
        {
            wf.GitSha = gitSha;
            _db.Workflows.Add(wf);
            _db.Tasks.AddRange(compiled.Tasks);
            _db.TaskEdges.AddRange(compiled.Edges);
            return;
        }

        existing.Name = wf.Name;
        existing.Status = wf.Status;
        existing.Mode = wf.Mode;
        existing.Target = wf.Target;
        existing.Version = wf.Version;
        existing.TriggerJson = wf.TriggerJson;
        existing.GitSha = gitSha;
        existing.CompiledAt = now;
        existing.UpdatedAt = now;

        var existingTasks = existing.Tasks.ToDictionary(t => t.Id);

        // Preload archived rows for every task that is not in the active
        // definition in one query, instead of one SELECT per new task.
        var missingIds = compiled.Tasks
            .Where(t => !existingTasks.ContainsKey(t.Id))
            .Select(t => t.Id)
            .ToList();
        var archivedById = missingIds.Count == 0
            ? new Dictionary<Guid, WorkflowTask>()
            : (await _db.Tasks
                    .IgnoreQueryFilters()
                    .Where(t => missingIds.Contains(t.Id))
                    .ToListAsync(ct))
                .ToDictionary(t => t.Id);

        foreach (var task in compiled.Tasks)
        {
            if (existingTasks.TryGetValue(task.Id, out var existingTask))
            {
                ApplyTask(existingTask, task);
            }
            else
            {
                // The task may exist as an archived row (removed, then re-added with the
                // same id). Re-activate it in place instead of inserting a duplicate row
                // (deterministic id => primary-key collision). Historical TaskRuns stay
                // linked and future runs use this definition again.
                if (archivedById.TryGetValue(task.Id, out var archived))
                {
                    ApplyTask(archived, task);
                    archived.ArchivedAt = null;
                }
                else
                {
                    _db.Tasks.Add(task);
                }
            }
        }

        // Tasks removed from the definition no longer wipe their historical TaskRuns.
        // Each stale task is archived (preserving run history for audit/debug) and
        // excluded from active-definition queries via the WorkflowTask global query
        // filter. The row is kept so its TaskRuns' Task->TaskRun Restrict FK holds.
        var newIds = compiled.Tasks.Select(t => t.Id).ToHashSet();
        foreach (var stale in existing.Tasks.Where(t => !newIds.Contains(t.Id)).ToList())
        {
            stale.ArchivedAt = now;
        }

        SyncEdges(existing, compiled.Edges);
    }

    /// <summary>
    /// Reconciles the persisted edges with the compiled ones by deterministic id:
    /// removed edges are deleted, new edges added, unchanged edges left alone.
    /// Legacy script workflows compile no edges, so any stale node edges from a
    /// previous node-mode version of the same file are dropped.
    /// </summary>
    private void SyncEdges(Workflow existing, IReadOnlyList<WorkflowTaskEdge> compiled)
    {
        var desired = compiled.ToDictionary(e => e.Id);
        var current = existing.TaskEdges.ToDictionary(e => e.Id);

        foreach (var edge in existing.TaskEdges.Where(e => !desired.ContainsKey(e.Id)).ToList())
            _db.TaskEdges.Remove(edge);

        foreach (var edge in compiled)
        {
            if (current.TryGetValue(edge.Id, out var tracked))
                tracked.IsLoopBack = edge.IsLoopBack; // classification can change with the graph
            else
                _db.TaskEdges.Add(edge);
        }
    }

    private static void ApplyTask(WorkflowTask target, WorkflowTask source)
    {
        target.Key = source.Key;
        target.ParentId = source.ParentId;
        target.NextId = source.NextId;
        target.OnErrorId = source.OnErrorId;
        target.Language = source.Language;
        target.Mode = source.Mode;
        target.EntryJson = source.EntryJson;
        target.EnvJson = source.EnvJson;
        target.Server = source.Server;
        target.NodeType = source.NodeType;
        target.NodeVersion = source.NodeVersion;
        target.ParametersJson = source.ParametersJson;
        target.CredentialsJson = source.CredentialsJson;
        target.RetryJson = source.RetryJson;
        target.ContinueOnFail = source.ContinueOnFail;
        target.RunMode = source.RunMode;
        target.Order = source.Order;
    }

    /// <summary>
    /// Deletes a workflow definition whose source file is gone. The schema has
    /// <c>WorkflowRun->Workflow</c> (Restrict) and <c>TaskRun->Task</c>
    /// (Restrict), so an executed workflow can't be removed outright: its
    /// <c>WorkflowRun</c> rows are deleted first (which cascades their
    /// <c>TaskRun</c> attempts via <c>RunId</c>), then the workflow and its task
    /// tree (which cascades via <c>WorkflowId</c>).
    /// </summary>
    private async Task RemoveWorkflowAsync(Workflow workflow, CancellationToken ct)
    {
        await _db.WorkflowRuns
            .Where(r => r.WorkflowId == workflow.Id)
            .ExecuteDeleteAsync(ct);

        _db.Workflows.Remove(workflow);
    }
}