Einleitung

Im TeilWir haben einen schönen visuellen Editor gebaut, aber unsere Workflows laufen nur dann, wenn wir sie manuell auslösen. In diesem letzten Beitrag machen wir die Arbeitsabläufe mit Hangfire für ,:

• Geplante Ausführung Führe Workflows auf einem Zeitplan aus
• API-Abfrage Überwachen Sie externe APIs und triggern Sie Änderungen
• Staatliche Verwaltung Verfolgen Sie Triggerzustände über Exekutionen hinweg
• Dashboard Überwachen Sie alle Hintergrundarbeiten

Warum Hangfire?

Hangfire ist perfekt für unsere Bedürfnisse, weil es

• Speichert Jobs in unserer bestehenden PostgreSQL Datenbank
• Bietet ein integriertes Dashboard
• Unterstützt wiederkehrende Jobs
• Hat automatische Wiederholung Logik
• Waagen horizontal

Das Trigger State Model

Erste let-'s verstehen unsere Trigger Staat Entity -( , wir haben dies bereits in Teil erstellt.

[Table("workflow_trigger_states")]
public class WorkflowTriggerStateEntity
{
    public int Id { get; set; }
    public int WorkflowDefinitionId { get; set; }

    // Type: "Schedule", "ApiPoll", "Webhook"
    public string TriggerType { get; set; } = string.Empty;

    // Configuration as JSON
    public string ConfigJson { get; set; } = "{}";

    // Current state as JSON (stores last poll time, content hash, etc.)
    public string StateJson { get; set; } = "{}";

    public bool IsEnabled { get; set; } = true;
    public DateTime? LastCheckedAt { get; set; }
    public DateTime? LastFiredAt { get; set; }
    public int FireCount { get; set; } = 0;
    public string? LastError { get; set; }
}

Diese Entity verfolgt alles über einen Workflow-Trigger.

• Als es zuletzt lief
• Was seine Konfiguration ist
• Welcher Zustand es ist, ' ist in .( für stattliche Trigger.
• Fehler, die aufgetreten sind

Geplante Workflows

Konfigurationsmodell

public class ScheduleTriggerConfig
{
    public string IntervalType { get; set; } = "minutes"; // minutes, hours, days
    public int IntervalValue { get; set; } = 60;
    public Dictionary<string, object>? InputData { get; set; }
}

Der Planerjob

public class WorkflowSchedulerJob
{
    private readonly MostlylucidDbContext _context;
    private readonly WorkflowExecutionService _executionService;
    private readonly ILogger<WorkflowSchedulerJob> _logger;

    [AutomaticRetry(Attempts = 3)]
    public async Task ExecuteScheduledWorkflowsAsync()
    {
        _logger.LogInformation("Checking for scheduled workflows");

        // Get all enabled schedule triggers
        var triggers = await _context.WorkflowTriggerStates
            .Include(t => t.WorkflowDefinition)
            .Where(t => t.IsEnabled && t.TriggerType == "Schedule")
            .ToListAsync();

        foreach (var trigger in triggers)
        {
            try
            {
                var config = JsonSerializer.Deserialize<ScheduleTriggerConfig>(
                    trigger.ConfigJson);

                if (config == null) continue;

                // Check if it's time to run
                if (!ShouldRunScheduledWorkflow(trigger, config))
                    continue;

                _logger.LogInformation(
                    "Executing scheduled workflow {WorkflowId}",
                    trigger.WorkflowDefinition.WorkflowId);

                // Execute the workflow
                await _executionService.ExecuteWorkflowAsync(
                    trigger.WorkflowDefinition.WorkflowId,
                    config.InputData,
                    "Scheduler");

                // Update trigger state
                trigger.LastCheckedAt = DateTime.UtcNow;
                trigger.LastFiredAt = DateTime.UtcNow;
                trigger.FireCount++;

                var state = JsonSerializer.Deserialize<Dictionary<string, object>>(
                                trigger.StateJson) ?? new();
                state["lastRun"] = DateTime.UtcNow.ToString("O");
                trigger.StateJson = JsonSerializer.Serialize(state);

                await _context.SaveChangesAsync();
            }
            catch (Exception ex)
            {
                _logger.LogError(ex,
                    "Error executing scheduled workflow {TriggerId}",
                    trigger.Id);
                trigger.LastError = ex.Message;
                await _context.SaveChangesAsync();
            }
        }
    }

    private bool ShouldRunScheduledWorkflow(
        WorkflowTriggerStateEntity trigger,
        ScheduleTriggerConfig config)
    {
        // First run?
        if (!trigger.LastFiredAt.HasValue)
            return true;

        var timeSinceLastRun = DateTime.UtcNow - trigger.LastFiredAt.Value;

        return config.IntervalType.ToLower() switch
        {
            "minutes" => timeSinceLastRun.TotalMinutes >= config.IntervalValue,
            "hours" => timeSinceLastRun.TotalHours >= config.IntervalValue,
            "days" => timeSinceLastRun.TotalDays >= config.IntervalValue,
            _ => false
        };
    }
}

Wie es funktioniert

1. Jede Minute Anrufe aus dem Hangfire ExecuteScheduledWorkflowsAsync()
2. Wir suchen nach aktivierten Zeitplanauslösern
3. Prüfen Sie für jeden Trigger, ob genügend Zeit vergangen ist
4. Wenn ja, führen Sie den Workflow aus
5. Aktualisieren des Triggerzustands mit der letzten Laufzeit

API-Abfrage

API-Abfrage ist interessanter.Wir überwachen externe APIs und lösen Workflows aus, wenn sich der Inhalt ändert.

Konfigurationsmodell

public class ApiPollTriggerConfig
{
    public string Url { get; set; } = string.Empty;
    public int IntervalSeconds { get; set; } = 300; // 5 minutes
    public bool AlwaysTrigger { get; set; } = false;
    public Dictionary<string, string>? Headers { get; set; }
}

Die Umfragearbeit

[AutomaticRetry(Attempts = 3)]
public async Task PollApiTriggersAsync()
{
    _logger.LogInformation("Polling API triggers");

    var triggers = await _context.WorkflowTriggerStates
        .Include(t => t.WorkflowDefinition)
        .Where(t => t.IsEnabled && t.TriggerType == "ApiPoll")
        .ToListAsync();

    foreach (var trigger in triggers)
    {
        try
        {
            var config = JsonSerializer.Deserialize<ApiPollTriggerConfig>(
                trigger.ConfigJson);

            if (config == null) continue;

            // Check if it's time to poll
            if (trigger.LastCheckedAt.HasValue)
            {
                var timeSinceLastCheck = DateTime.UtcNow - trigger.LastCheckedAt.Value;
                if (timeSinceLastCheck.TotalSeconds < config.IntervalSeconds)
                    continue;
            }

            _logger.LogInformation("Polling API for workflow {WorkflowId}",
                trigger.WorkflowDefinition.WorkflowId);

            // Poll the API
            using var httpClient = new HttpClient();
            var response = await httpClient.GetAsync(config.Url);
            var content = await response.Content.ReadAsStringAsync();

            // Get previous state
            var state = JsonSerializer.Deserialize<Dictionary<string, object>>(
                            trigger.StateJson) ?? new();

            var previousHash = state.GetValueOrDefault("contentHash")?.ToString();
            var currentHash = ComputeHash(content);

            // Has content changed?
            if (previousHash != currentHash || config.AlwaysTrigger)
            {
                _logger.LogInformation(
                    "API content changed, triggering workflow {WorkflowId}",
                    trigger.WorkflowDefinition.WorkflowId);

                // Pass response as input to workflow
                var inputData = new Dictionary<string, object>
                {
                    ["apiResponse"] = content,
                    ["statusCode"] = (int)response.StatusCode,
                    ["previousHash"] = previousHash ?? string.Empty,
                    ["currentHash"] = currentHash
                };

                // Execute the workflow
                await _executionService.ExecuteWorkflowAsync(
                    trigger.WorkflowDefinition.WorkflowId,
                    inputData,
                    $"ApiPoll:{config.Url}");

                trigger.LastFiredAt = DateTime.UtcNow;
                trigger.FireCount++;

                // Update state
                state["contentHash"] = currentHash;
                state["lastContent"] = content.Length > 1000
                    ? content.Substring(0, 1000)
                    : content;
                state["lastPoll"] = DateTime.UtcNow.ToString("O");
            }

            trigger.LastCheckedAt = DateTime.UtcNow;
            trigger.StateJson = JsonSerializer.Serialize(state);
            trigger.LastError = null;

            await _context.SaveChangesAsync();
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error polling API trigger {TriggerId}",
                trigger.Id);
            trigger.LastError = ex.Message;
            trigger.LastCheckedAt = DateTime.UtcNow;
            await _context.SaveChangesAsync();
        }
    }
}

private string ComputeHash(string content)
{
    using var sha256 = System.Security.Cryptography.SHA256.Create();
    var bytes = System.Text.Encoding.UTF8.GetBytes(content);
    var hash = sha256.ComputeHash(bytes);
    return Convert.ToBase64String(hash);
}

Wie es funktioniert

1. Überprüfen Sie jede Minute alle API-Abfrageauslöser
2. Prüfen Sie für jeden Trigger, ob seit der letzten Umfrage genügend Zeit vergangen ist.
3. Umfrage der konfigurierten URL
4. Berechnung eines Hashs des Antwortinhalts
5. Vergleich mit dem im Zustand gespeicherten Hash
6. Wenn geändert AlwaysTrigger ist true.),-Ausführung des Workflows
7. Übergeben Sie die API-Antwort als Eingabedaten an den Workflow
8. Zustand mit neuem Hash aktualisieren

Fallbeispiel verwenden

Überwachen Sie GitHub-Veröffentlichungen

{
  "triggerType": "ApiPoll",
  "config": {
    "url": "https://api.github.com/repos/dotnet/aspnetcore/releases/latest",
    "intervalSeconds": 3600,
    "alwaysTrigger": false
  }
}

Diese Umfragen die GitHub API jede Stunde Wenn eine neue Veröffentlichung veröffentlicht wird, ändert sich der Inhalt Hash., und der Workflow führt mit den Release-Daten aus

Registrierung von Hangfire-Jobs

In Ihrem Program.cs oder Konfiguration des Starts

// Add Hangfire services
builder.Services.AddHangfire(config =>
{
    config.UsePostgreSqlStorage(
        builder.Configuration.GetConnectionString("DefaultConnection"));
});

builder.Services.AddHangfireServer();

// Register our job
builder.Services.AddScoped<WorkflowSchedulerJob>();

Dann, nachdem die App gestartet hat, registrieren Sie wiederkehrende Jobs.

app.UseHangfireDashboard("/hangfire");

// Register recurring jobs
RecurringJob.AddOrUpdate<WorkflowSchedulerJob>(
    "scheduled-workflows",
    job => job.ExecuteScheduledWorkflowsAsync(),
    Cron.Minutely);

RecurringJob.AddOrUpdate<WorkflowSchedulerJob>(
    "api-poll-triggers",
    job => job.PollApiTriggersAsync(),
    Cron.Minutely);

Das Hangfire-Dashboard

Hangfire enthält ein eingebautes --Dashboard, das unter /hangfire:

• ArbeitsplätzeAlle Warteschlangen ansehen und abgeschlossene Jobs anzeigen
• Wiederkehrende JobsVerwalten Sie unsere Workflow-Scheduler
• WiederholungenAuswählen und erneut versuchen fehlgeschlagene Jobs
• ServerÜberwachung von Hangfire-Servern

Sicherung des Dashboards

app.UseHangfireDashboard("/hangfire", new DashboardOptions
{
    Authorization = new[]
    {
        new HangfireAuthorizationFilter()
    }
});

public class HangfireAuthorizationFilter : IDashboardAuthorizationFilter
{
    public bool Authorize(DashboardContext context)
    {
        var httpContext = context.GetHttpContext();

        // Only allow authenticated users
        return httpContext.User.Identity?.IsAuthenticated == true;
    }
}

Steuerung von Triggern über UI

Lassen Sie 's hinzufügen Benutzeroberfläche für das Erstellen und Verwalten von Triggern

[HttpPost("workflow/{id}/triggers")]
public async Task<IActionResult> CreateTrigger(
    string id,
    [FromBody] TriggerCreateRequest request)
{
    var workflow = await _context.WorkflowDefinitions
        .FirstOrDefaultAsync(w => w.WorkflowId == id);

    if (workflow == null)
        return NotFound();

    var trigger = new WorkflowTriggerStateEntity
    {
        WorkflowDefinitionId = workflow.Id,
        TriggerType = request.Type,
        ConfigJson = JsonSerializer.Serialize(request.Config),
        StateJson = "{}",
        IsEnabled = true
    };

    await _context.WorkflowTriggerStates.AddAsync(trigger);
    await _context.SaveChangesAsync();

    return Json(new { success = true, triggerId = trigger.Id });
}

public class TriggerCreateRequest
{
    public string Type { get; set; } = string.Empty; // Schedule, ApiPoll
    public object Config { get; set; } = new();
}

UI-Bauteil

<div class="card bg-base-100 shadow-xl">
    <div class="card-body">
        <h2 class="card-title">⏰ Add Trigger</h2>

        <div class="form-control">
            <label class="label">Trigger Type</label>
            <select class="select select-bordered" x-model="triggerType">
                <option value="Schedule">Schedule</option>
                <option value="ApiPoll">API Poll</option>
            </select>
        </div>

        <!-- Schedule Config -->
        <template x-if="triggerType === 'Schedule'">
            <div class="space-y-4">
                <div class="form-control">
                    <label class="label">Interval</label>
                    <div class="flex gap-2">
                        <input type="number"
                               x-model="scheduleConfig.intervalValue"
                               class="input input-bordered flex-1" />
                        <select x-model="scheduleConfig.intervalType"
                                class="select select-bordered">
                            <option value="minutes">Minutes</option>
                            <option value="hours">Hours</option>
                            <option value="days">Days</option>
                        </select>
                    </div>
                </div>
            </div>
        </template>

        <!-- API Poll Config -->
        <template x-if="triggerType === 'ApiPoll'">
            <div class="space-y-4">
                <div class="form-control">
                    <label class="label">API URL</label>
                    <input type="url"
                           x-model="apiConfig.url"
                           class="input input-bordered"
                           placeholder="https://api.example.com/data" />
                </div>

                <div class="form-control">
                    <label class="label">Poll Interval (seconds)</label>
                    <input type="number"
                           x-model="apiConfig.intervalSeconds"
                           class="input input-bordered"
                           value="300" />
                </div>
            </div>
        </template>

        <button @click="createTrigger()" class="btn btn-primary mt-4">
            Create Trigger
        </button>
    </div>
</div>

Real--.World Beispiel Workflow

Lassen Sie uns einen kompletten automatisierten Workflow erstellen, der

1. Umfragen der GitHub API für neue Versionen
2. Überprüft, ob die Version neuer ist als das was wir gesehen haben
3. Protokolliert eine Nachricht
4. Eine E-Mail senden könnte... an Slack, , usw.

Schritt Erstellen Sie den Workflow

{
  "name": "GitHub Release Monitor",
  "startNodeId": "parse-data",
  "nodes": [
    {
      "id": "parse-data",
      "type": "Transform",
      "name": "Extract Version",
      "inputs": {
        "operation": "json_parse",
        "data": "{{apiResponse}}"
      }
    },
    {
      "id": "log-release",
      "type": "Log",
      "name": "Log New Release",
      "inputs": {
        "message": "New release: {{tag_name}} - {{name}}",
        "level": "info"
      }
    }
  ],
  "connections": [
    {
      "sourceNodeId": "parse-data",
      "targetNodeId": "log-release"
    }
  ]
}

Schritt Erstellen Sie den API-Poll Trigger

{
  "type": "ApiPoll",
  "config": {
    "url": "https://api.github.com/repos/dotnet/aspnetcore/releases/latest",
    "intervalSeconds": 3600
  }
}

Jetzt ,. jede Stunde ,. Hangfire wird :.

1. Umfrage der GitHub API
2. Vergleichen Sie den Inhalt Hash mit vorheriger Umfrage
3. Wenn geändert, führen Sie den Workflow aus
4. Der Workflow parsiert den JSON und protokolliert die Release-Info

Überwachung und Beobachtungsfähigkeit

Protokollierung

Alle Workflow-Ausführungen werden protokolliert

_logger.LogInformation(
    "Workflow {WorkflowId} execution {ExecutionId} completed in {Duration}ms with status {Status}",
    execution.WorkflowId,
    execution.Id,
    execution.DurationMs,
    execution.Status);

Metrik

Wir können Prometheus Metriken hinzufügen

private static readonly Counter WorkflowExecutions = Metrics
    .CreateCounter("workflow_executions_total",
        "Total workflow executions",
        new CounterConfiguration
        {
            LabelNames = new[] { "workflow_id", "status" }
        });

// In execution service
WorkflowExecutions
    .WithLabels(workflow.Id, execution.Status.ToString())
    .Inc();

Ausschreibungen

Alarme einrichten für:

• fehlgeschlagene Arbeitsabläufe
• Workflows dauern zu lange
• API-Abfragefehler
• Auslöser, die im erwarteten Zeitrahmen abgefeuert haben

Leistungsbeurteilungen

Datenbank laden

Bei vielen Workflows kann häufig Umfragen die Datenbanklast signifikant sein.

Lösungsabfragen für Batch

// Instead of querying per trigger
var triggers = await _context.WorkflowTriggerStates
    .Include(t => t.WorkflowDefinition)
    .Where(t => t.IsEnabled && t.TriggerType == "ApiPoll")
    .AsNoTracking() // Read-only
    .ToListAsync();

API-Rate begrenzen

Wenn externe APIs abgefragt werden

Lösung: Exponielles Backoff

catch (HttpRequestException ex) when (ex.StatusCode == HttpStatusCode.TooManyRequests)
{
    // Back off
    var retryAfter = response.Headers.RetryAfter?.Delta ?? TimeSpan.FromMinutes(5);
    state["backoffUntil"] = DateTime.UtcNow.Add(retryAfter).ToString("O");
}

Erweiterte Funktionen

Bedingte Auslöser

Trigger nur, wenn bestimmte Bedingungen erfüllt sind

public class ConditionalTriggerConfig : ApiPollTriggerConfig
{
    public string? Condition { get; set; } // e.g., "{{stars}} > 1000"
}

Abhängigkeiten auslösen

Kettenauslöser eines Arbeitsablaufs eines Abschlusses von ' löst ein anderes aus

// After workflow completes
if (execution.Status == WorkflowExecutionStatus.Completed)
{
    var dependentTriggers = await _context.WorkflowTriggerStates
        .Where(t => t.TriggerType == "WorkflowComplete" &&
                    t.ConfigJson.Contains(execution.WorkflowId))
        .ToListAsync();

    foreach (var trigger in dependentTriggers)
    {
        await _executionService.ExecuteWorkflowAsync(
            trigger.WorkflowDefinition.WorkflowId,
            execution.OutputData,
            $"Triggered by {execution.WorkflowId}");
    }
}

Prüfen von Hangfire-Jobs

Einheit testen Sie Ihre Jobs

[Fact]
public async Task ExecuteScheduledWorkflows_ShouldExecuteWhenIntervalPassed()
{
    // Arrange
    var mockContext = CreateMockContext();
    var mockExecutionService = new Mock<IWorkflowExecutionService>();
    var job = new WorkflowSchedulerJob(mockContext.Object,
        mockExecutionService.Object, Mock.Of<ILogger>());

    // Act
    await job.ExecuteScheduledWorkflowsAsync();

    // Assert
    mockExecutionService.Verify(s => s.ExecuteWorkflowAsync(
        It.IsAny<string>(),
        It.IsAny<Dictionary<string, object>>(),
        "Scheduler",
        It.IsAny<CancellationToken>()), Times.Once);
}

Schlußfolgerung

Wir haben ein komplettes Automatisierungssystem aufgebaut. Unsere Workflows können jetzt.

✅ Auf Zeitpläne laufen Tägliche oder benutzerdefinierte Intervalle ✅ Umfrage-APIs Überwachen Sie externe Dienste auf Änderungen ✅ Gleiszustand Erinnere dich an das, was wir vorher gesehen haben ✅ Auto---Retry Behandeln Sie vorübergehende Ausfälle ✅ Überwachung Dashboard für alle Jobs ✅ Skala Hängfeuerhandgriffe Lastausgleich

Die vollständige Serie

Wir haben ein Enterprise-Workflow-System von scratch-:

• TeilEinführung und Architektur
• TeilKern-Workflow-Engine
• TeilVisualer Workflow-Editor
• TeilDie Integration der Hangfire-Einbindung von ,( , diesem Beitrag ,) ,

Du hast jetzt ,:'

• Eine leistungsstarke Workflow-Engine
• Ein schöner visueller Editor
• Automatisierte Ausführung
• API-Überwachung
• Vollständige Beobachtbarkeit

Was ist das nächste?

Mögliche Verbesserungen

• WebhooksAuslösung von Workflows über HTTP-Endpunkte
• E-Mail-NodesE-Mails von Workflows senden
• DatenbankknotenAbfrage-Datenbanken
• AI-KnotenMit LLMs integrieren
• Sub---WorkflowsZusammensetzen von Workflows
• Workflow-MarktplatzWorkflow-Vorlagen teilen

Quellencode

Alle Codes sind im Repository verfügbar.

• Hangfire-Jobs Mostlylucid.SchedulerService/Jobs/
• Workflow-Modelle - : Mostlylucid.Workflow.Shared/
• Workflow-Engine - : Mostlylucid.Workflow.Engine/

Vielen Dank, dass Sie dieser Serie gefolgt sind.