Introduzione

In Parte 3, abbiamo costruito un bellissimo editore visivo. Ma i nostri flussi di lavoro funzionano solo quando li scatenamo manualmenteM SK2 In questo post finale , noiMSC4 faremo i flussi lavorativi veramente autonomi usando Hangfire perMST5

• L'esecuzione programmata - Condurre i flussi di lavoro su un programma
• I sondaggi API - Monitorare gli API esterni e attivare i cambiamenti
• Gestione statale - Stanze di attivazione tra le eseguzioni
• La lavagna - Monitorare tutti i lavori di fondo

Perché Hangfire?

Hangfire è perfetto per i nostri bisogni perché

• Gestisce lavori nella nostra base di dati PostgreSQL.
• Fornisce un dashboard - costruito.
• Sostene lavori ricorrenti
• Ha una logica di prova automatica.
• Scala in modo orizzontale

Il modello di stato Trigger

Prima , lasciare ' capire la nostra entità di stato di attivazione МSK2 l'abbiamo già creata nella parte

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

Questa entità traccia tutto su un trigger del flusso di lavoro:

• Quando è andata l'ultima volta?
• Qual è la sua configurazione?
• Qual è lo stato di ' in (per i getti statici)
• Ci sono stati errori.

Workflow programmati

Modello di configurazione

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

Il lavoro del programmatore

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

Come funziona:

1. Ogni minuto, Le telefonate di Hangfire ExecuteScheduledWorkflowsAsync()
2. Ci chiediamo se ci sono trigger programmati.
3. Per ogni trigger, controllare se è passato abbastanza tempo.
4. Se sì, eseguire il flusso di lavoro
5. aggiornare lo stato di attivazione con il tempo di esecuzione ultimo

I sondaggi API

Il sondaggio delle API è più interessante - monitoriamo le API esterne e scatenamo i flussi di lavoro quando il contenuto cambia!

Modello di configurazione

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

Il lavoro del sondaggio

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

Come funziona:

1. Ogni minuto, controllare tutti i rilevatori di sondaggio API
2. Per ogni trigger, si deve controllare se è passato abbastanza tempo dall'ultimo sondaggio.
3. Pollare l'URL configurata
4. Computare un hash del contenuto della risposta.
5. Confrontare con l'hash precedente immagazzinato nello stato
6. Se cambiato (o AlwaysTrigger è vero), eseguire il flusso di lavoro
7. Passare la risposta API come dati input al flusso di lavoro.
8. aggiornare lo stato con un nuovo hash

Usare un esempio di caso

Monitorare le rilasciazioni di GitHub:

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

Questo sondaggio analizza l'API di GitHub ogni ora. Quando viene pubblicata una nuova versione, cambia il hash del contenutoM SK2 e il flusso di lavoro si svolge con i dati della versione !

Registrare i lavori di Hangfire

Nella vostra casa. Program.cs o la configurazione di startup:

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

builder.Services.AddHangfireServer();

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

Poi, dopo che l'app iniziaM SK1 registrare lavori ricorrenti:

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

La lavagna di Hangfire

Hangfire include un dashboard - costruito accessibile a /hangfire:

• I posti di lavoro: Vedere tutti i lavori in fila, processoM SK2 e compiuti
• Posti ricurrenti: Gestire i nostri programmatori del flusso di lavoro
• Ritorni: View and retry failed jobs
• I server: Monitor Hangfire server

Assicurare il Dashboard

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

Gestire i Trigger attraverso l'UI

Lasciate's aggiungere l'interfaccia utente per creare e gestire i getti.

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

Componente UI

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

Facciamo costruire un intero flusso di lavoro automatizzato che

1. Polls the GitHub API for new releases
2. Controlla se la versione è più nuova di quella che abbiamo visto.
3. Registra un messaggio
4. (Potrebbe mandare un e-mail, posta a SlackM SK2 ecc.

Step 1: Creare il 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"
    }
  ]
}

Step 2: Creare l'API Poll Trigger

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

Ora, ogni oraM SK1 Hangfire sarà:

1. Pollare l'API di GitHub
2. Confrontare l'hash del contenuto con il sondaggio precedente.
3. Se cambiato, eseguire il flusso di lavoro.
4. Il flusso di lavoro analizza il JSON e registra le informazioni sul rilascio.

Monitoring e osservabilità

Loggazione

Tutte le eseguzioni del flusso di lavoro sono registrate:

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

Metrica

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

Alerti

Inserire l'allarme per:

• Flussi di lavoro falliti (Status ==Facciolo)
• I flussi di lavoro prendono troppo tempo.
• I fallimenti del sondaggio API
• Triggeri che non sono stati attivati nel tempo previsto.

Considerazioni di performance

carico della base di dati

Con molti flussi di lavoro che fanno frequenti sondaggi, il carico della base di dati può essere significativo:

Soluzione: Queries di lotto

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

Limitazione del tasso dell'API

Quando sondare API esterni:

Soluzione: Backoff esponenziale

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

I vantaggi avanzati

Triggeri Condizionali

Triggere solo se alcune condizioni sono soddisfatte:

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

Dependenze di Trigger

La catena attiva - uno flusso di lavoro' l' completazione attiva un altroM 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}");
    }
}

Testare i lavori di Hangfire

Unità per testare i vostri lavori:

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

Conclusione

Abbiamo costruito un sistema completo di automatizzazione.

✅ Funzionare su schemi - HourlyM SK1 daily, o intervalli personalizzati ✅ API di sondaggio - Controllare i servizi esterni per i cambiamenti ✅ Lo stato della traccia - Ricordate quello che abbiamo visto prima ✅ Auto-retry - Controlli i fallimenti transitori ✅ Monitore - Dashboard per tutti i lavori ✅ Scala - Hangfire fa bilanciare il carico

La serie completa

Noi'abbiamo costruito un'azienda-un sistema di flusso di lavoro a livello da zeroM SK2

• Parte 1: Introduzione e architettura
• Parte 2: Motore di flusso di lavoro centrale
• Parte 3: Editore del flusso di lavoro visivo
• Parte 4: Integrazione di Hangfire (Questo postM SK2

Ora avete:

• Un potente motore del flusso di lavoro.
• Un bellissimo editore visivo.
• Eseguimento automatizzato
• Monitoraggio dell'API
• Visibilità completa

Cosa's successivo?

Possibili miglioramenti:

• Webhooks: Triggera i flussi di lavoro attraverso i punti finali HTTP
• Nodi email: Invia email dai flussi di lavoro
• Nodi di base dati: Database di domande
• Nodi AI: Integrare con i LLM
• Workflow Sub-: Compose workflows together
• Il mercato del flusso di lavoro: Template di lavoro condiviso

Il codice sorgente

Tutti i codici sono disponibili nel riposito:

• Hangfire lavori: Mostlylucid.SchedulerService/Jobs/
• modelli di flusso di lavoro: Mostlylucid.Workflow.Shared/
• Motore di lavoro: Mostlylucid.Workflow.Engine/

Grazie per aver seguito questa serie! Happy workflow building! M SK2