Introduction

À l'intérieur Partie 3, Nous avons construit un beau rédacteur visuel . Mais nos flux de travail ne fonctionnent que lorsque nous les déclenchons manuellement M SK2 Dans ce dernier article , nous MSC4 ferons des flux de travaux vraiment autonomes en utilisant Hangfire pour

• Exécution prévue - Faire fonctionner les flux de travail sur un calendrier
• Enquête API - Surveiller les API externes et déclencher les changements
• Gestion de l'État - Les états de déclenchement d’exécution à travers la piste
• Tableau de bord - Surveiller tous les emplois de fond

Pourquoi Hangfire?

Hangfire est parfait pour nos besoins parce qu’il

• Stockage des emplois dans notre base de données PostgreSQL existante
• Offre un tableau de bord -in
• Soutenir les emplois récurrents
• Logique d'essai automatique
• Évalue horizontalement

Le modèle d'État de déclenchement

Premièrement, nous devons permettre à l’entité déclencheur , de comprendre notre entité d’État déclencheur ( Nous l’avons déjà créée dans la partie

[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; }
}

Cette entité suit tout ce qui touche un déclencheur de flux de travail:

• Quand elle a eu lieu pour la dernière fois
• Quelle est sa configuration
• Quel est le degré d’état des déclencheurs stables?
• Toute erreur qui a eu lieu

Les flux de travail prévus

Modèle de configuration

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

Le travail de l'échéancier

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

Comment il fonctionne:

1. Every minute, Hangfire calls ExecuteScheduledWorkflowsAsync()
2. Nous demandons des déclencheurs de calendrier activés
3. Pour chaque déclencheur, vérifiez si suffisamment de temps est passé
4. Si oui, exécuter le flux de travail
5. Mise à jour de l'état du déclencheur avec le dernier temps d'exécution

Enquête sur l'API

Le sondage des API est plus intéressant - nous surveillons les API externes et déclenchons les flux de travail lorsque le contenu change!

Modèle de configuration

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

Le travail du sondage

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

Comment il fonctionne:

1. Chaque minute, vérifiez tous les déclencheurs de sondage API
2. Pour chaque déclencheur, vérifiez s’il y a eu suffisamment de temps depuis le dernier sondage
3. Poller l'URL configurée
4. Calculer un hash du contenu de la réponse
5. Comparaison avec le hash précédent entreposé dans l'état
6. Si modifié (ou AlwaysTrigger est vrai), exécuter le flux de travail
7. Passer la réponse API en tant que données d'entrée au flux de travail
8. Mise à jour de l'état avec un nouveau hash

Utiliser un exemple de cas

Surveillance des mises à jour de GitHub:

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

Ce sondage est effectué chaque heure sur l’API de GitHub.

Enregistrement des emplois Hangfire

Dans votre Program.cs ou configuration de démarrage:

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

builder.Services.AddHangfireServer();

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

Ensuite, après le début de l'applicationM SK1 enregistrer les emplois répétés:

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

Le tableau de bord Hangfire

Hangfire comprend un tableau de bord - intégré accessible à /hangfire:

• Emplois: Voir tous les postes à la queueM SK1 traitement , et emplois terminés
• Travaux récurrents: Gérer nos horaires de travail
• Retraits: Voir et essayer les tâches manquantes
• Les serveurs: Surveillance des serveurs Hangfire

Assurer le tableau de bord

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

Gestion des déclencheurs par l'intermédiaire de la UI

Let's add UI for creating and managing triggers:

[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();
}

Component de l'interface

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

Exemple de travail Real-World

Let's construire un flux de travail automatisé complet qui

1. Enquête sur l'API de GitHub pour les nouvelles versions
2. Vérifie si la version est plus récente que ce que nous avons vu
3. Enregistre un message
4. (Possibilité d'envoyer un courriel, post à SlackM SK2 etc .)

Étape 1: Créer le processus de travail

{
  "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"
    }
  ]
}

Étape 2: Créer l'API Poll Trigger

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

Aujourd’hui, chaque heure , Hangfire va:

1. Poller l'API de GitHub
2. Comparer le hash du contenu avec un sondage antérieur
3. Si modifiée,, exécutez le flux de travail
4. Le flux de travail analyse le JSON et enregistre l'information sur la mise à jour.

Surveillance et observabilité

Enregistrement

Toutes les exécutions du flux de travail sont enregistrées:

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

Metriques

Nous pouvons ajouter Prometheus metrics:

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();

Alertes

Établir des alertes pour:

• Les flux de travail ont échoué (Status == Échoué)
• Les flux de travail prennent trop longtemps
• Défaillances de sondage API
• Triggers qui n’ont pas été activés dans le cadre de temps attendu

Considérations relatives au rendement

Charge de la base de données

Avec de nombreux flux de travail faisant des sondages fréquemment, la charge de base de données peut être significative,

Solution: Demandes de lots

// 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();

Limitation du taux de l'API

Lors du sondage des API externes:

Solution: Retour exponentiel

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

Caractéristiques avancées

Triggers conditionnels

Seulement déclencher si certaines conditions sont remplies:

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

Dépendences de déclenchement

Chaîne déclenche - un flux de travail' l'achèvement déclenchait un autreM SK2

// 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}");
    }
}

Évaluation des emplois Hangfire

Unité d’essai de vos emplois:

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

Conclusion

Nous avons construit un système d’automatisation complet.

✅ Exécuter sur les calendriers - horaire , quotidienM SK2 ou intervalles personnalisés ✅ API de sondage - Surveiller les services externes en cas de changement ✅ État de la piste - Rappelez-vous ce que nous avons déjà vu ✅ Auto-retry - Gérer les pannes transitoires ✅ Contrôleur - tableau de bord pour tous les postes ✅ Établir une échelle - Hangfire permet d'équilibrer la charge

La série complète

Nous ' avons construit une entreprise - un système de flux de travail d’échelon à partir du bas

• Partie 1: Introduction et architecture
• Partie 2: moteur de travail principal
• Partie 3: Éditeur du flux de travail visuel
• Partie 4: Intégration de Hangfire (le présent articleM SK2

Vous avez maintenant:

• Un puissant moteur de travail
• Un beau rédacteur visuel
• Exécution automatisée
• Surveillance de l'API
• Observabilité totale

Qu’est-ce que's Suivant?

Améliorations possibles:

• Webhooks: Trigger les flux de travail par l'intermédiaire des points d'entrée HTTP
• Nodes électroniques: Envoyer les courriels des flux de travail
• Nodes de base de données: Bases de données de demande
• Nodes d'AI: Intégrer avec les LLM
• Sub-flows de travail: Composer les flux de travail ensemble
• Marché du travail: Modèles de travail partagés

Code de source

Tous les codes sont disponibles dans le répertoire:

• Emplois de Hangfire: Mostlylucid.SchedulerService/Jobs/
• Modèles de travail: Mostlylucid.Workflow.Shared/
• moteur de travail: Mostlylucid.Workflow.Engine/

Merci d'avoir suivi cette série! Happy workflow building ! 🎉