المقدمة

في الجزء 3, قمنا ببناء محرر بصري جميل. لكن عمليتنا تعمل فقط عندما نشغلها لاسيًا . في هذا المقطع الأخيرM SK3 سنجعل عمليتنا ذاتية حقيقية باستخدام هانغفيير لـ:

• تنفيذ مخطط - تنفيذ عمليات العمل على جدول زمني
• تصويت الAPI - تراقب APIs الخارجية وتقوم بتحريك التغييرات
• إدارة الدولة - حالة جرس المسار عبر تنفيذات
• لوحة الجدول - تراقب جميع الوظائف الخلفية

لماذا هانغفيير?

"انغ فاير" مثالي لمتطلباتنا لأنه

• يحفظ الوظائف في قاعدة بياناتنا PostgreSQL الموجودة
• يقدم لوحة مفاتيح مبنية -
• يدعم الوظائف المتكررة
• لديه منطق إعادة المحاولة تلقائيا
• يقيس بشكل افقي

نموذج الدولة المحفزة

أولاً , دعونا ' نفهم هياكل الدولة المحفزة لدينا | ( | لقد قمنا بالفعل بخلق هذا في الجزء |2):

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

هذه الكيان تتبع كل شيء حول مشغل تدفق العمل:

• عندما تم تشغيله أخيرا
• ما هي تركيبه
• ما هي الحالة 's في (for stateful triggers
• أي أخطاء حدثت

تدفقات العمل المخططة

نموذج الإعداد

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

وظيفة المخطط

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

كيف يعمل

1. كل دقيقة, مكالمات "هنجفيير" ExecuteScheduledWorkflowsAsync()
2. نحن نسأل عن محفزات جدول فعالة
3. لكل مفتاح , تأكد من أنه تم مرور الوقت الكافي
4. إذا كان نعم, تنفيذ تدفق العمل
5. تحديث حالة الاستثارة مع وقت تشغيل الأخير

استطلاع للرأي لAPI

إستطلاع الAPI أكثر إثارة للاهتمام - نراقب الAPI الخارجية ونقوم بتحفيز عمليات العمل عندما يتغير المحتوى!

نموذج الإعداد

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

وظيفة التصويت

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

كيف يعمل

1. كل دقيقة, تأكد من جميع محفزات إستطلاع API
2. لل trigger , إختبر ما إذا كان هناك وقت كافي منذ الاستطلاع الأخير
3. إجراء إستطلاع على الURL المبرمج
4. حساب هش محتويات الاستجابة
5. مقارنة مع الهش السابق المخزن في الحالة
6. إذا تغيرت (or AlwaysTrigger هو صحيح), تنفيذ تدفق العمل
7. تنقل استجابة الAPI كبيانات إدخال إلى تدفق العمل
8. تحديث الحالة ب Hash جديد

استخدم مثال حالة

مراقبة نشرات GitHub:

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

يقوم هذا باستطلاع واجهة GitHub كل ساعة. عندما يتم نشر إصدار جديد, يتغير هاش المحتوىM SK2 وينفذ التدفق العملي مع بيانات إصداره

تسجيل وظيفة "هنجفيير"

في Program.cs أو إعداد البداية:

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

builder.Services.AddHangfireServer();

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

ثم , بعد أن يبدأ التطبيق, تسجيل الوظائف المتكررةM SK2

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

لوحة الاشتعال

Hangfire يحتوي على لوحة مفاتيح مبنية - متاحة على /hangfire:

• الوظائف: شاهدوا جميع الوظائف المصطفة, المعالجة , وتلك التي تم إنجازها
• الوظائف المتكررة: إدارة مخططات عمليتنا
• التقاعد: رؤية ومحاولة أخرى لم تنجح في الوظائف
• المخدمات: مراقبة серверات "هنجفيير"

حماية لوحة الأدلة

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

إدارة المؤشرات عن طريق UI

دعونا نضيف UI لخلق وإدارة المحفزات

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

مكون واجهة المستخدم

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

عمل عملي حقيقي -نموذج عالمي

دعونا نبني عملية عمل آلية كاملة

1. استطلاع لAPI GitHub للإصدارات الجديدة
2. يراقب ما إذا كان الإصدار أقدم مما رأيناه
3. يسجل رسالة
4. (يمكن إرسال بريد إلكتروني,رسال رسالة إلى SlackM SK2 وهلم جرا .)

الخطوة 1: إنشاء عملية العمل

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

الخطوة 2: إنشاء محرك إستطلاع API

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

الآن , كل ساعة , سطل النار :

1. إستطلاع واجهة GitHub
2. قارن "هاش المحتوى" مع الاستطلاع السابق
3. إذا قمت بتغيير , بتنفيذ عملية العمل
4. عملية العمل تقوم بتحليل JSON وتسجل معلومات الإصدار

المراقبة والملاحظة

السجل

جميع عمليات تدفق العمل تم تسجيلها:

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

القياس

يمكننا إضافة قياسات بروميثيوس:

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

التحذيرات

إعداد التحذيرات ل:

• فشلت عملية العمل (وضع ==فشلت)
• تدفقات العمل تستغرق وقتا طويلا جدا
• فشل API في التصويت
• المحفزات التي لم تطلق في الإطار الزمني المتوقع

وجهات النظر في الأداء

تحميل قاعدة بيانات

مع الكثير من التدفقات العاملة تصويت باستمرار, شحنة قاعدة البيانات يمكن أن تكون كبيرة:

الحل: أسئلة بطاقة

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

عند إستطلاع APIs الخارجية:

الحل: تراجع عرضي

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

الميزات المتقدمة

المحفزات conditional

فقط إذا كانت بعض الشروط ملائمة:

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

إختلافات تحفيز

ت trigger السلسلة - عملية عمل واحدة ' تمويلها ت trigger الأخرى

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

اختبار وظيفة "هنجفيير"

اختبار وحدة عملك:

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

الاستنتاجات

لقد بنينا نظام اتومات كامل

✅ تشغيل على الجدول - ساعة, يوميًاM SK2 أو وقفات مخصصة ✅ تطبيقات إستطلاع - تراقب الخدمات الخارجية للتغيير ✅ حالة المسار - تذكر ما رأيناه من قبل ✅ Auto-retry - تخطي الفشل přechodي ✅ المراقب - لوحة المفاتيح لكل الوظائف ✅ قياس - يتحكم في توازن الشحن

سلسلة كاملة

لقد بنينا "مسك0" شركة "مسک1" نظام تدفق العمل من الصفر

• الجزء 1: مقدمة وفن العمارة
• الجزء 2: محرك سير العمل الأساسي
• الجزء 3: محرر لتدفق العمل البصري
• الجزء 4: إدماج هانغفيير (هذا пост)

الآن لديكم

• محرك عمل قوي
• محرر بصري جميل
• تنفيذ تلقائي
• مراقبة الAPI
• قابلية الملاحظة الكاملة

ماذا's التالي?

تحسينات ممكنة:

• ويبهوكس: تمكّن من تحريك تدفقات العمل من خلال نقاط النهاية HTTP
• نقاط البريد الإلكتروني: أرسل رسائل بريدية من مسارات العمل
• العقد في قاعدة البيانات: قاعدة بيانات الأسئلة
• عقدة الذكاء الصناعي: تدمج مع LLMs
• فرعي-تدفقات العمل: إنشاء عمليات عمل معاً
• سوق العمل: شراكة قواميس عملية العمل

الرمز المصدر

كل البرمجة متوفرة في المخزن:

• وظائف "هنج فايير": Mostlylucid.SchedulerService/Jobs/
• نماذج تدفق العمل: Mostlylucid.Workflow.Shared/
• محرك تدفق العمل: Mostlylucid.Workflow.Engine/

شكراً لمتابعة هذا السلسلة! بناء عملية عمل سعيدة ! 🎉