Job Engine — Developer Guide

GB5 Enterprise Scheduler & Queue Processor · .NET 9 · Quartz 3.13 · Dapper · FastEndpoints

Architecture Overview

The Job Engine is a standalone GB5 microservice (port 5400) responsible for two concerns:

• Scheduled execution — Quartz.NET clustered persistent store drives cron / interval / one-time jobs, calling HTTP endpoints (MWEBSERVICE) or Dapr topics.
• Async message queue — Persistent TJOBQUEUE table replaces MSMQ. Module-specific BackgroundService subclasses claim, process, and retry batches.

Module Structure

GB5Solution/JobEngine/
├── JobEngineDAL/
│   ├── DTOs/          JobDefineDTO, SchedulerDTO, JobExecutionDTO, JobQueueItemDTO, JobDashboardDTO
│   ├── Query/         JobDefineQB, SchedulerQB, JobExecutionQB, JobQueueQB      ← SQL constants only
│   ├── Interfaces/    IJobDefineDAL, ISchedulerDAL, IJobExecutionDAL, IJobQueueDAL
│   └── Implementations/
├── JobEngineBLL/
│   ├── Interfaces/    IJobDefineBLL, ISchedulerBLL, IJobExecutionBLL, IJobQueueBLL, IQuartzSyncService
│   └── Implementations/
└── JobEngineSL/
    ├── EndPoints/     17 FastEndpoints (GET/POST/PUT/DELETE)
    ├── Hubs/          JobEngineHub.cs, IJobEngineClient.cs
    ├── Services/      QuartzJobExecutor, QuartzSyncService, OutBoxPollerService
    ├── Migrations/    JobEngine_Migration.sql
    └── Program.cs

Database Schema

Existing tables (extended via ALTER TABLE)

New tables (created by migration)

Migration script: JobEngineSL/Migrations/JobEngine_Migration.sql. Run once; idempotent (ALTER TABLE only adds columns that don't exist).

Quartz.NET Integration

QuartzSyncService — job lifecycle management

Singleton registered in DI. BLL calls it via IQuartzSyncService so BLL never depends on Quartz types directly.

QuartzJobExecutor — per-execution IJob

A new instance is created for every execution. Execution flow:

API Endpoints

All endpoints follow the GB5 FastEndpoint pattern. Registered base path: /JobEngine. Gateway prefix: /je/.

Job Definition

Scheduler

Execution History

Queue + DLQ

SignalR Events

Hub: JobEngineHub mapped to /hubs/jobengine. All events broadcast to group jobengine:client:{ClientId}. Transport: LongPolling (WebSocket blocked by custom Login header).

Queue Processor (BaseJobQueueProcessor)

Module-specific processors inherit BaseJobQueueProcessor (in GB5Shared/QueueReader/). They specify which MessageType values they own.

Implemented processors

Atomic claim SQL

The claim query uses UPDATE TOP(@BatchSize) … OUTPUT INSERTED.* WITH (ROWLOCK, READPAST) to guarantee no two processor instances ever claim the same row, even when running in a Kubernetes cluster.

Retry backoff schedule

Adding a New Scheduled Job

Two paths depending on who owns the job:

Path A — Admin UI (via Job Define screen)

1. Create a TSCHEDULER row via Schedule Manager screen (type: CRON / INTERVAL / ONE_TIME).
2. In Job Define, pick the scheduler, the target MWEBSERVICE URL, and optional URI parameter.
3. Save → BLL calls IQuartzSyncService.SyncJobAsync() which immediately registers the trigger in Quartz.

Path B — Module-driven registration

BLL calls IJobDefineBLL.RegisterJobAsync() which does an upsert by (SourceModule, SourceObjectId, TenantId):

await _jobDefineBLL.RegisterJobAsync(new JobDefineDTO
{
    JobName       = $"Report:{dto.ReportName}",
    JobCategory   = "REPORT",
    SourceModule  = "REPORTS",
    SourceObjectId = dto.ReportId,
    WebServiceId  = dto.ReportWebServiceId,
    // Scheduler is created inline by RegisterJobAsync if not provided
    TenantId      = login.ClientId,
    Status        = 1
}, login, ct);

Adding a New Queue Message Handler

Implement IJobQueueHandler in the relevant module's BLL or a shared library:

public class MyNewHandler : IJobQueueHandler
{
    public string MessageType => "MY_MESSAGE_TYPE";

    public async Task HandleAsync(string payload, LoginDTO login, CancellationToken ct)
    {
        var dto = JsonSerializer.Deserialize<MyPayloadDTO>(payload)!;
        // ... business logic ...
    }
}

Register in the module's Program.cs:

builder.Services.AddScoped<IJobQueueHandler, MyNewHandler>();

Then add "MY_MESSAGE_TYPE" to the relevant processor's MessageTypes list. The base processor resolves handlers by type at runtime.

Processor Registration in Program.cs

// MMSL/Program.cs
builder.Services.AddHostedService<MMJobQueueProcessor>();

// PayRollSL/Program.cs
builder.Services.AddHostedService<PAYJobQueueProcessor>();

ActionProcessor Integration

The ActionProcessor system (in FrameworkBLL/ActionProcessor/) handles event-driven action delivery — email, SMS, webhook, and SignalR notifications. It is complementary to (not part of) the Job Engine.

Integration point: when QuartzJobExecutor completes a job, it can publish a SchedulerTaskDTO to the Scheduler.Ready RabbitMQ queue. SchedulerActionGeneratorBLL picks this up, evaluates MEVENTACTION rules (JSONPath conditions), and queues matching delivery actions in TEVENTACTIONRUN + TACTIONOUTBOX.

OpenTelemetry Instrumentation

Every BLL save method must follow this pattern:

public async Task<string> SaveJobAsync(JobDefineDTO dto, LoginDTO login, CancellationToken ct)
{
    GB5Trace.Step("validate-job-define", new { dto.JobId });
    _validation.NotEmpty(dto.JobName, nameof(dto.JobName));

    var isNew = dto.JobId == 0;
    GB5Trace.Step("save-job-define", new { dto.JobId, isNew });
    var savedId = await _dal.SaveJobAsync(dto, login, ct).ConfigureAwait(false);

    GB5Trace.Step("sync-quartz", new { savedId });
    await _quartzSync.SyncJobAsync(savedId, ct).ConfigureAwait(false);

    return isNew
        ? $"{SuccessResponse.SaveSuccessMessage} {savedId}"
        : SuccessResponse.UpdateSuccess;
}
catch (Exception ex)
{
    GB5Trace.MarkFailed("save-job-define-failed", ex);
    _logger.LogError(ex, "SaveJob failed for {JobName}", dto.JobName);
    throw;
}

New Job / Handler Checklist