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;

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

    public async Task<SyncResult> SyncAsync(string tenantId, string repoName, CancellationToken ct)
    {
        // In Forgejo-backed mode, pull the latest changes before reading files.
        if (_sourceFactory?.IsForgejoBacked == true && _sourceFactory.Forgejo != null)
        {
            _logger.LogDebug("Pulling latest workflow files from Forgejo for tenant {TenantId}", tenantId);
            var login = await _sourceFactory.ResolveForgejoLoginAsync(tenantId, ct);
            if (!string.IsNullOrEmpty(login))
                await _sourceFactory.Forgejo.PullAsync(tenantId, login, ct);
        }

        var state = _source.GetState();
        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)
            .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);
            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);
        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.
                var archived = await _db.Tasks
                    .IgnoreQueryFilters()
                    .FirstOrDefaultAsync(t => t.Id == task.Id, ct);
                if (archived != null)
                {
                    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;
        }
    }

    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.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);
    }
}