# Aufbau eines Workflow-Systems mit HTMX und ASP-.-NET Core Teil -, Hangfire Integration und Automatisierung

<!--category-- ASP.NET, Hangfire, Workflow, Background Jobs -->
<datetime class="hidden">2025-01-15T18:00</datetime>

## Einleitung

Im [Teil](/blog/workflowsystem-part3-visual-editor)Wir 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

[TOC]

## 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.

```csharp
[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

```csharp
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

```csharp
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

```csharp
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

```csharp
[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**

```json
{
  "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

```csharp
// 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.

```csharp
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ätze**Alle Warteschlangen ansehen und abgeschlossene Jobs anzeigen
- **Wiederkehrende Jobs**Verwalten Sie unsere Workflow-Scheduler
- **Wiederholungen**Auswählen und erneut versuchen fehlgeschlagene Jobs
- **Server**Überwachung von Hangfire-Servern

### Sicherung des Dashboards

```csharp
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

```csharp
[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

```html
<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

```json
{
  "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

```json
{
  "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

```csharp
_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

```csharp
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**

```csharp
// 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**

```csharp
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

```csharp
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

```csharp
// 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

```csharp
[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-:

- **Teil**Einführung und Architektur
- **Teil**Kern-Workflow-Engine
- **Teil**Visualer Workflow-Editor
- **Teil**Die 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

- **Webhooks**Auslösung von Workflows über HTTP-Endpunkte
- **E-Mail-Nodes**E-Mails von Workflows senden
- **Datenbankknoten**Abfrage-Datenbanken
- **AI-Knoten**Mit LLMs integrieren
- **Sub---Workflows**Zusammensetzen von Workflows
- **Workflow-Marktplatz**Workflow-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.