Background Processing
Arthix.Worker — RabbitMQ consumer, scheduled action poller, and SignalR relay. Runs as a Windows Service.
When to use this module
- Create a new RabbitMQ queue for async processing.
- Workflow node requires a delay.
- Worker completes a task and needs to send a notification.
- Implement a new background job type.
Architecture Diagram
API Server Worker Service
┌────────────────────┐ ┌───────────────────────────┐
│ RabbitMqPublisher │───Queues───→ │ RabbitMqConsumerWorker │
│ (Infrastructure) │ │ (single BackgroundService) │
└────────────────────┘ │ │ │
│ ├─ WorkflowQueueHandler │
│ ├─ WhatsAppMessageHandler │
│ ├─ WhatsAppTemplateHandler│
│ ├─ MetaWebhookHandler │
│ ├─ ContactImportHandler │
│ └─ SyncMetaTemplateHandler│
├───────────────────────────┤
│ ScheduledActionWorker │
│ polls every 30 seconds │
├───────────────────────────┤
│ WorkerSignalRClient │
│ → API NotificationHub │
└───────────────────────────┘
Queue Handlers
| Handler | Queue Name | What It Does |
|---|---|---|
WhatsAppMessageQueueHandler | whatsapp-message | Calls Meta API to send message |
WhatsAppTemplateQueueHandler | whatsapp-template | Template CRUD on Meta API |
MetaWebhookQueueHandler | meta-webhook | Process webhook events |
ContactImportQueueHandler | contact-import | Parse CSV + create contacts |
SyncMetaTemplateQueueHandler | sync-meta-template | Pull templates from Meta |
WorkflowQueueHandler | workflow | Execute workflow node |
IQueueMessageHandler Interface
C#
public interface IQueueMessageHandler
{
string QueueName { get; }
Task HandleAsync(string message, IServiceScope scope, CancellationToken ct);
}Scheduled Actions
Scheduled Action Worker polls every 30 seconds:
Pseudocode
Loop:
SELECT * FROM ScheduledActions WHERE Status = Pending AND ScheduledFor <= UtcNow
For each action:
Mark as Processing
Dispatch based on type:
Workflow → resume workflow from delay node
WorkflowMessageWait → timeout message wait listener
Mark as Completed (or Failed)| Type | Created By | Dispatches To |
|---|---|---|
Workflow | Delay node | WorkflowProccessService (resume from next node) |
WorkflowMessageWait | Button wait | WorkflowProccessService (timeout handler) |
How to Extend
How to add a new queue:
- Add method to
RabbitMqPublisher.cs→PublishYourTaskAsync() - Create
YourQueueHandler.csimplementingIQueueMessageHandler - In
Worker/Program.cs:builder.Services.AddSingleton<IQueueMessageHandler, YourQueueHandler>(); - RabbitMqConsumerWorker auto-discovers — no other changes needed.
How to add a new scheduled action:
- Add value to
ScheduledActionTypeEnum - Add case in
ScheduledActionWorker.DispatchActionAsync()
Configuration
JSON
{
"RabbitMq": {
"HostName": "localhost",
"Port": 5672,
"UserName": "guest",
"Password": "guest",
"VirtualHost": "/"
},
"SignalR": {
"HubUrl": "https://localhost:5001/notificationhub",
"RetryDelayMs": 5000
}
}