Quartz.NET with Wolverine

builder.UseWolverine(opts =>
{
    // Handlers live in this assembly. Setting it explicitly rather than letting Wolverine walk the
    // stack is what keeps discovery working under a test runner and under a second host in the same
    // process (JasperFx/wolverine#3776, #3778).
    opts.ApplicationAssembly = typeof(OrderPlaced).Assembly;

    // Part 3 sends raw bytes to an endpoint by name, and a name is the whole of what it can address —
    // there is no live message for Wolverine to route on. A local queue keeps that free of a broker.
    opts.PublishMessage<ArchiveOrders>().ToLocalQueue(Part3RawMessageData.EndpointName);

    // Part 7: Wolverine's own recurring schedule, registered here rather than inside AddQuartz because
    // it is Wolverine's. Since 6.34 this is what most of part 1 would otherwise be reaching for.
    Part7WolverineSchedules.Register(opts, options.ExpiryCron);

    if (options.HasDatabase)
    {
        // The outbox, the inbox and the node table part 5's agent needs. Quartz's own tables go in the
        // same database, because part 6's single transaction cannot span two servers.
        opts.PersistMessagesWithPostgresql(options.PostgresConnectionString!);
    }
});

builder.Services.AddQuartz(q =>
{
    Part1RecurringPublishing.Register(q, options.ReconciliationCron);
    Part4TunedLatency.Register(q);

    if (options.HasDatabase)
    {
        q.UsePersistentStore(store =>
        {
            store.UsePostgres(options.PostgresConnectionString!);
            store.UseSystemTextJsonSerializer();

            // Weasel creates and migrates Quartz's tables with Wolverine's: at startup, and from
            // db-apply and resources setup. Quartz.Weasel.PostgreSQL; ProvisionSchema() without it.
            store.UseWeaselForPostgres();

            // Part 6 throws without this, rather than silently scheduling outside the caller's
            // transaction.
            store.ConfigureStore(o => o.AcceptEnlistedTransactions = true);
        });
    }

    // Nothing else: the in-memory store is what a scheduler falls back to, so UseInMemoryStore() would
    // only restate the default.
});

public static void Register(WolverineOptions opts, string cron)
{
    // Cronos' grammar, not Quartz's. The zone is the schedule's own, so a deployment that means
    // "03:15 local" says so here rather than hoping the host agrees — the same decision part 1
    // makes with InTimeZone, and one of the few this feature and a Quartz trigger both let you make.
    CronSchedule schedule = new(cron, TimeZoneInfo.Utc);

    // The factory is handed the occurrence time, which is this feature's answer to reading
    // context.ScheduledFireTimeUtc rather than the clock: the message describes the window the
    // schedule says it is for. Without a name the schedule is named for its message type, and
    // ScheduleRecurring<T>(cron) is the whole registration for a message with a parameterless
    // constructor.
    opts.Schedules.ScheduleRecurring(ScheduleName, schedule, occurrence => new ExpireUnpaidOrders(occurrence));
}

public sealed class ReconciliationJob : IJob<ReconciliationWindow>
{
    private readonly IMessageBus bus;
    private readonly ILogger<ReconciliationJob> logger;

    public ReconciliationJob(IMessageBus bus, ILogger<ReconciliationJob> logger)
    {
        this.bus = bus;
        this.logger = logger;
    }

    public async ValueTask Execute(
        IJobExecutionContext context,
        ReconciliationWindow input,
        CancellationToken cancellationToken = default)
    {
        // The scheduler's own clock, not DateTimeOffset.UtcNow: a trigger that misfired and is firing
        // late still reports the time it was scheduled for, which is the window the run is about.
        DateTimeOffset to = context.ScheduledFireTimeUtc ?? context.FireTimeUtc;

        RunReconciliation message = new(to - input.Length, to);
        await bus.PublishAsync(message);

        logger.LogInformation("Published {Message} for the window ending {To:O}", nameof(RunReconciliation), to);
    }
}

q.ScheduleJob<ReconciliationJob>(trigger => trigger
    .WithIdentity("reconciliation", "recurring")
    .WithCronSchedule(cron, x => x
        // The expression is read in this zone, so a deployment that means "03:00 local" says
        // so here rather than hoping the host agrees.
        .InTimeZone(TimeZoneInfo.Utc)
        // What happens when the process was down at 03:00. DoNothing skips to the next
        // firing, which is what part 7's schedule does and all it does; FireAndProceed
        // publishes one catch-up message, which is the choice Wolverine's own schedules do
        // not offer.
        .WithMisfireInstruction(CronTriggerMisfireInstruction.DoNothing))
    .UsingInput(new ReconciliationWindow(TimeSpan.FromDays(1))));

public static class OrderPlacedHandler
{
    public static async Task Handle(OrderPlaced message, IScheduler scheduler, CancellationToken cancellationToken)
    {
        ScheduledOneOffJob scheduled = await scheduler.ScheduleJob<PaymentReminderJob, PaymentReminder>(
            new PaymentReminder(message.OrderId, message.Amount),
            ExampleOptions.Current.ReminderDelay,
            new OneOffJobOptions { Group = OrderGroup.For(message.OrderId) },
            cancellationToken);

        // The call answers with the trigger's key and the time the store says it will first fire, so
        // "scheduled for" is what will happen rather than what was asked for.
        Ledger.Record(Events.ReminderScheduled, $"{scheduled.TriggerKey} at {scheduled.FirstFireTimeUtc:u}");
    }
}

public static class OrderPaidHandler
{
    public static async Task Handle(OrderPaid message, IScheduler scheduler, CancellationToken cancellationToken)
    {
        // The whole cancellation, in one store operation. The matcher is evaluated where the triggers
        // are, so no key list round-trips through this process and there is no window in which a
        // trigger listed a moment ago fires before it can be removed. What comes back is the keys that
        // were actually withdrawn, which is how the caller learns whether it beat the firing.
        List<TriggerKey> cancelled = await scheduler.UnscheduleJobs(
            GroupMatcher<TriggerKey>.GroupEquals(OrderGroup.For(message.OrderId)),
            cancellationToken);

        Ledger.Record(Events.RemindersCancelled, $"{cancelled.Count} for {message.OrderId}");
    }
}

public static async ValueTask<TriggerKey> ScheduleSend<TMessage>(
    IScheduler scheduler,
    IWolverineRuntime runtime,
    TMessage message,
    TimeSpan delay,
    CancellationToken cancellationToken = default) where TMessage : notnull
{
    // Serialized here, at the moment the decision was made, rather than at fire time.
    DeferredEnvelope envelope = new(
        EndpointName,
        typeof(TMessage).ToMessageTypeName(),
        runtime.Options.DefaultSerializer.WriteMessage(message));

    ScheduledOneOffJob scheduled = await scheduler.ScheduleJob<DeferredEnvelopeJob, DeferredEnvelope>(
        envelope,
        delay,
        new OneOffJobOptions { Group = "deferred-envelopes" },
        cancellationToken);

    return scheduled.TriggerKey;
}

public async ValueTask Execute(
    IJobExecutionContext context,
    DeferredEnvelope input,
    CancellationToken cancellationToken = default)
{
    IDestinationEndpoint endpoint = bus.EndpointFor(input.EndpointName);

    await endpoint.SendRawMessageAsync(input.Data, configure: envelope =>
    {
        // The stored name rather than SetMessageType<T>(): the type this envelope describes is
        // whatever was serialized, which this job has no static knowledge of.
        envelope.MessageType = input.MessageTypeName;

        // Setting Destination is not optional, and Wolverine 6.35.0 does not do it for you.
        // DestinationEndpoint.SendRawMessageAsync assigns Sender but leaves Destination null, and
        // Executor.ExecuteAsync logs both success and failure through envelope.Destination!, so a
        // raw message that is handled perfectly still ends the pipeline with a
        // NullReferenceException out of the logging call. One line here avoids it.
        envelope.Destination = endpoint.Uri;
    });
}

q.ConfigureScheduler(options =>
{
    // Default 30 s. Only affects how quickly this node notices triggers it did not schedule
    // itself, so it is a clustering setting, not a latency setting.
    options.IdleWaitTime = TimeSpan.FromSeconds(10);

    // Default automatic: the pool on a persistent store, clustered or not, 1 in memory.
    // Must not exceed ThreadPoolOptions.MaxConcurrency, which defaults to 10.
    options.MaxBatchSize = 10;

    // Default TimeSpan.Zero, which batches only triggers already due. This also batches
    // triggers due up to half a second after the first, and fires them that much early.
    options.BatchTriggerAcquisitionFireAheadTimeWindow = TimeSpan.FromMilliseconds(500);
});

builder.Services.AddQuartzHostedService(hosted =>
{
    hosted.AutoStart = false;
    hosted.WaitForJobsToComplete = true;
});

public sealed class SchedulerStarter : IHostedService
{
    private readonly ISchedulerFactory schedulerFactory;
    private readonly ILogger<SchedulerStarter> logger;

    public SchedulerStarter(ISchedulerFactory schedulerFactory, ILogger<SchedulerStarter> logger)
    {
        this.schedulerFactory = schedulerFactory;
        this.logger = logger;
    }

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        IScheduler scheduler = await schedulerFactory.GetScheduler(cancellationToken);
        await scheduler.Start(cancellationToken);

        logger.LogInformation("Scheduler '{Name}' started after the Wolverine runtime", scheduler.SchedulerName);
        Ledger.Record(Events.SchedulerStartedByWolverine, "IHostedService ordered after UseWolverine");
    }

    // Nothing to do: the Quartz hosted service shuts the scheduler down whether or not it started it.
    public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
}

protected override async Task startAsync(CancellationToken cancellationToken)
{
    started = await schedulerFactory.GetScheduler(cancellationToken);
    await started.Start(cancellationToken);

    logger.LogInformation("Scheduler '{Name}' started on this node by Wolverine", started.SchedulerName);
    Ledger.Record(Events.SchedulerStartedByWolverine, "Wolverine SingularAgent");
}

protected override async Task stopAsync(CancellationToken cancellationToken)
{
    // The scheduler the agent started, not a fresh ISchedulerFactory.GetScheduler(): on host
    // shutdown Wolverine stops its agents after the Quartz hosted service has already shut the
    // scheduler down, and asking the factory for it again throws rather than handing back the
    // shut-down instance. Holding the reference and checking Status keeps the stop quiet.
    if (started is null || started.Status is SchedulerStatus.ShuttingDown or SchedulerStatus.Shutdown)
    {
        return;
    }

    // Standby rather than Shutdown: the agent may be re-assigned to this node later, and a
    // shut-down scheduler cannot be started again in the same container.
    await started.Standby(cancellationToken);
}

public static async Task Handle(
    ApproveRefund message,
    MessageContext context,
    IWolverineRuntime runtime,
    IScheduler scheduler,
    CancellationToken cancellationToken)
{
    IMessageDatabase database = (IMessageDatabase) runtime.Storage;

    await using NpgsqlConnection connection = new(ExampleOptions.Current.PostgresConnectionString!);
    await connection.OpenAsync(cancellationToken);
    await using DbTransaction transaction = await connection.BeginTransactionAsync(cancellationToken);

    // Wolverine's outgoing envelopes are now written into this transaction rather than sent
    // immediately. This is the manual form of what [Transactional] does for a Marten or EF Core
    // application.
    await context.EnlistInOutboxAsync(new DatabaseEnvelopeTransaction(database, transaction));

    // 1. the application's own state
    await using (NpgsqlCommand command = new(
        "insert into refunds (order_id, amount) values (@order_id, @amount)",
        connection,
        (NpgsqlTransaction) transaction))
    {
        command.Parameters.AddWithValue("order_id", message.OrderId);
        command.Parameters.AddWithValue("amount", message.Amount);
        await command.ExecuteNonQueryAsync(cancellationToken);
    }

    // 2. a message that must not be sent unless the row above survives
    await context.PublishAsync(new SendPaymentReminder(message.OrderId, message.Amount));

    // 3. the trigger, in the same transaction as both, with the commit inside the scope so the
    // scheduler is signalled once the trigger is visible to it
    using (scheduler.EnlistTransaction(transaction))
    {
        await scheduler.ScheduleJob<PaymentReminderJob, PaymentReminder>(
            new PaymentReminder(message.OrderId, message.Amount),
            TimeSpan.FromDays(7),
            new OneOffJobOptions { Group = OrderGroup.For(message.OrderId) },
            cancellationToken);

        await transaction.CommitAsync(cancellationToken);
    }

    // Releases the envelopes the outbox held back. Nothing left the process before the commit.
    await context.FlushOutgoingMessagesAsync();

    Ledger.Record(Events.RefundApprovedInTransaction, message.OrderId);