301 lines
13 KiB
C#
301 lines
13 KiB
C#
using System.Data;
|
|
using Microsoft.Extensions.Options;
|
|
using Npgsql;
|
|
using Tiku.Application.Jobs;
|
|
using Tiku.Application.PlatformBilling;
|
|
using Tiku.Application.Security;
|
|
using Tiku.Application.Tenancy;
|
|
using Tiku.Infrastructure.Observability;
|
|
using System.Diagnostics;
|
|
|
|
namespace Tiku.Worker;
|
|
|
|
public sealed class WorkerOptions
|
|
{
|
|
public const string SectionName = "Worker";
|
|
|
|
public bool Enabled { get; set; } = true;
|
|
public int JobPollSeconds { get; set; } = 2;
|
|
public int JobParallelism { get; set; } = 4;
|
|
public int JobBatchSize { get; set; } = 5;
|
|
|
|
public static bool BeValid(WorkerOptions options) =>
|
|
options.JobPollSeconds is >= 1 and <= 3600 &&
|
|
options.JobParallelism is >= 1 and <= 32 &&
|
|
options.JobBatchSize is >= 1 and <= 100;
|
|
}
|
|
|
|
internal interface IPeriodicProcessorLock
|
|
{
|
|
Task<IAsyncDisposable?> TryAcquireAsync(string processor, CancellationToken cancellationToken);
|
|
}
|
|
|
|
internal sealed class WorkerStateReporter(NpgsqlDataSource dataSource)
|
|
{
|
|
private readonly string workerId = $"{Environment.MachineName}:{Environment.ProcessId}";
|
|
private readonly DateTimeOffset startedAt = DateTimeOffset.UtcNow;
|
|
|
|
public Task StartedAsync(string processor, CancellationToken cancellationToken) =>
|
|
UpsertAsync(processor, true, null, cancellationToken);
|
|
|
|
public Task CompletedAsync(string processor, Exception? error, CancellationToken cancellationToken) =>
|
|
UpsertAsync(processor, false, error?.Message, cancellationToken);
|
|
|
|
private async Task UpsertAsync(
|
|
string processor,
|
|
bool running,
|
|
string? error,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var now = DateTimeOffset.UtcNow;
|
|
await using var connection = await dataSource.OpenConnectionAsync(cancellationToken);
|
|
await using var command = connection.CreateCommand();
|
|
command.CommandText = """
|
|
INSERT INTO worker_heartbeats
|
|
(id, worker_id, processor, started_at, last_heartbeat_at, last_iteration_started_at,
|
|
last_iteration_completed_at, last_succeeded_at, last_error, is_running)
|
|
VALUES
|
|
(gen_random_uuid(), @worker_id, @processor, @started_at, @now,
|
|
CASE WHEN @running THEN @now ELSE NULL END,
|
|
CASE WHEN @running THEN NULL ELSE @now END,
|
|
CASE WHEN NOT @running AND @error IS NULL THEN @now ELSE NULL END,
|
|
@error, @running)
|
|
ON CONFLICT (worker_id, processor) DO UPDATE SET
|
|
last_heartbeat_at = EXCLUDED.last_heartbeat_at,
|
|
last_iteration_started_at = CASE WHEN EXCLUDED.is_running THEN EXCLUDED.last_heartbeat_at ELSE worker_heartbeats.last_iteration_started_at END,
|
|
last_iteration_completed_at = CASE WHEN EXCLUDED.is_running THEN worker_heartbeats.last_iteration_completed_at ELSE EXCLUDED.last_heartbeat_at END,
|
|
last_succeeded_at = CASE WHEN NOT EXCLUDED.is_running AND EXCLUDED.last_error IS NULL THEN EXCLUDED.last_heartbeat_at ELSE worker_heartbeats.last_succeeded_at END,
|
|
last_error = EXCLUDED.last_error,
|
|
is_running = EXCLUDED.is_running
|
|
""";
|
|
command.Parameters.AddWithValue("worker_id", workerId);
|
|
command.Parameters.AddWithValue("processor", processor);
|
|
command.Parameters.AddWithValue("started_at", startedAt);
|
|
command.Parameters.AddWithValue("now", now);
|
|
command.Parameters.AddWithValue("running", running);
|
|
command.Parameters.AddWithValue("error", (object?)error ?? DBNull.Value);
|
|
await command.ExecuteNonQueryAsync(cancellationToken);
|
|
}
|
|
}
|
|
|
|
internal sealed class PostgresPeriodicProcessorLock(NpgsqlDataSource dataSource) : IPeriodicProcessorLock
|
|
{
|
|
public async Task<IAsyncDisposable?> TryAcquireAsync(string processor, CancellationToken cancellationToken)
|
|
{
|
|
var connection = await dataSource.OpenConnectionAsync(cancellationToken);
|
|
await using var command = connection.CreateCommand();
|
|
command.CommandText = "SELECT pg_try_advisory_lock(hashtextextended(@processor, 0))";
|
|
command.Parameters.AddWithValue("processor", processor);
|
|
var acquired = (bool)(await command.ExecuteScalarAsync(cancellationToken) ?? false);
|
|
if (!acquired)
|
|
{
|
|
await connection.DisposeAsync();
|
|
return null;
|
|
}
|
|
|
|
return new AdvisoryLockLease(connection, processor);
|
|
}
|
|
|
|
private sealed class AdvisoryLockLease(NpgsqlConnection connection, string processor) : IAsyncDisposable
|
|
{
|
|
public async ValueTask DisposeAsync()
|
|
{
|
|
if (connection.State == ConnectionState.Open)
|
|
{
|
|
await using var command = connection.CreateCommand();
|
|
command.CommandText = "SELECT pg_advisory_unlock(hashtextextended(@processor, 0))";
|
|
command.Parameters.AddWithValue("processor", processor);
|
|
await command.ExecuteScalarAsync();
|
|
}
|
|
await connection.DisposeAsync();
|
|
}
|
|
}
|
|
}
|
|
|
|
internal abstract class PeriodicWorker(
|
|
ILogger logger,
|
|
IPeriodicProcessorLock processorLock,
|
|
WorkerStateReporter stateReporter,
|
|
string processorName,
|
|
TimeSpan interval,
|
|
bool enabled) : BackgroundService
|
|
{
|
|
protected abstract Task<int> ProcessAsync(CancellationToken cancellationToken);
|
|
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
if (!enabled)
|
|
{
|
|
return;
|
|
}
|
|
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
{
|
|
try
|
|
{
|
|
await using var lease = await processorLock.TryAcquireAsync(processorName, stoppingToken);
|
|
if (lease is not null)
|
|
{
|
|
var iterationTimestamp = Stopwatch.GetTimestamp();
|
|
await stateReporter.StartedAsync(processorName, stoppingToken);
|
|
try
|
|
{
|
|
var processed = await ProcessAsync(stoppingToken);
|
|
await stateReporter.CompletedAsync(processorName, null, stoppingToken);
|
|
WorkerTelemetry.RecordIteration(processorName, true, Stopwatch.GetElapsedTime(iterationTimestamp).TotalMilliseconds);
|
|
if (processed > 0)
|
|
{
|
|
logger.LogInformation("{Worker} processed {Count} items.", GetType().Name, processed);
|
|
}
|
|
}
|
|
catch (Exception exception)
|
|
{
|
|
await stateReporter.CompletedAsync(processorName, exception, CancellationToken.None);
|
|
WorkerTelemetry.RecordIteration(processorName, false, Stopwatch.GetElapsedTime(iterationTimestamp).TotalMilliseconds);
|
|
throw;
|
|
}
|
|
}
|
|
await Task.Delay(interval, stoppingToken);
|
|
}
|
|
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
|
{
|
|
break;
|
|
}
|
|
catch (Exception exception)
|
|
{
|
|
logger.LogError(exception, "{Worker} iteration failed.", GetType().Name);
|
|
try
|
|
{
|
|
await Task.Delay(interval, stoppingToken);
|
|
}
|
|
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
|
{
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
protected static void InitializeSystem(IServiceProvider services, string reason) =>
|
|
services.GetRequiredService<ITenantContextInitializer>().InitializeSystem(null, reason);
|
|
}
|
|
|
|
internal sealed class TenantDomainWorker(
|
|
IServiceScopeFactory scopeFactory,
|
|
IPeriodicProcessorLock processorLock,
|
|
WorkerStateReporter stateReporter,
|
|
IOptions<DomainLifecycleOptions> domainOptions,
|
|
IOptions<WorkerOptions> workerOptions,
|
|
ILogger<TenantDomainWorker> logger)
|
|
: PeriodicWorker(logger, processorLock, stateReporter, "tenant-domain-lifecycle",
|
|
TimeSpan.FromSeconds(Math.Clamp(domainOptions.Value.PollSeconds, 10, 3600)), workerOptions.Value.Enabled)
|
|
{
|
|
protected override async Task<int> ProcessAsync(CancellationToken cancellationToken)
|
|
{
|
|
await using var scope = scopeFactory.CreateAsyncScope();
|
|
InitializeSystem(scope.ServiceProvider, "Tenant domain DNS and TLS lifecycle worker");
|
|
return await scope.ServiceProvider.GetRequiredService<ITenantDomainLifecycleService>()
|
|
.ProcessPendingAsync(cancellationToken);
|
|
}
|
|
}
|
|
|
|
internal sealed class SaasSubscriptionWorker(
|
|
IServiceScopeFactory scopeFactory,
|
|
IPeriodicProcessorLock processorLock,
|
|
WorkerStateReporter stateReporter,
|
|
IOptions<WorkerOptions> options,
|
|
ILogger<SaasSubscriptionWorker> logger)
|
|
: PeriodicWorker(logger, processorLock, stateReporter, "saas-subscription-lifecycle", TimeSpan.FromSeconds(60), options.Value.Enabled)
|
|
{
|
|
protected override async Task<int> ProcessAsync(CancellationToken cancellationToken)
|
|
{
|
|
await using var scope = scopeFactory.CreateAsyncScope();
|
|
InitializeSystem(scope.ServiceProvider, "SaaS subscription lifecycle worker");
|
|
return await scope.ServiceProvider.GetRequiredService<ISaasSubscriptionLifecycleService>()
|
|
.ProcessDueAsync(cancellationToken: cancellationToken);
|
|
}
|
|
}
|
|
|
|
internal sealed class FeatureUsageWorker(
|
|
IServiceScopeFactory scopeFactory,
|
|
IPeriodicProcessorLock processorLock,
|
|
WorkerStateReporter stateReporter,
|
|
IOptions<FeatureUsageReconciliationOptions> featureOptions,
|
|
IOptions<WorkerOptions> workerOptions,
|
|
ILogger<FeatureUsageWorker> logger)
|
|
: PeriodicWorker(logger, processorLock, stateReporter, "feature-usage-reconciliation",
|
|
TimeSpan.FromMinutes(Math.Clamp(featureOptions.Value.IntervalMinutes, 1, 1440)), workerOptions.Value.Enabled)
|
|
{
|
|
protected override async Task<int> ProcessAsync(CancellationToken cancellationToken)
|
|
{
|
|
await using var scope = scopeFactory.CreateAsyncScope();
|
|
InitializeSystem(scope.ServiceProvider, "Tenant feature usage reconciliation worker");
|
|
return await scope.ServiceProvider.GetRequiredService<IFeatureUsageReconciliationService>()
|
|
.ProcessDueAsync(cancellationToken);
|
|
}
|
|
}
|
|
|
|
internal sealed class BackgroundJobsWorker(
|
|
IServiceScopeFactory scopeFactory,
|
|
IPeriodicProcessorLock processorLock,
|
|
WorkerStateReporter stateReporter,
|
|
IOptions<WorkerOptions> options,
|
|
ILogger<BackgroundJobsWorker> logger)
|
|
: PeriodicWorker(logger, processorLock, stateReporter, "background-jobs", TimeSpan.FromSeconds(options.Value.JobPollSeconds), options.Value.Enabled)
|
|
{
|
|
private readonly string workerId = $"{Environment.MachineName}:{Guid.NewGuid():N}";
|
|
private readonly int parallelism = options.Value.JobParallelism;
|
|
private readonly int batchSize = options.Value.JobBatchSize;
|
|
|
|
protected override async Task<int> ProcessAsync(CancellationToken cancellationToken)
|
|
{
|
|
var workers = Enumerable.Range(0, parallelism)
|
|
.Select(index => ProcessPartitionAsync(index, cancellationToken));
|
|
return (await Task.WhenAll(workers)).Sum();
|
|
}
|
|
|
|
private async Task<int> ProcessPartitionAsync(int index, CancellationToken cancellationToken)
|
|
{
|
|
await using var scope = scopeFactory.CreateAsyncScope();
|
|
InitializeSystem(scope.ServiceProvider, "Background job lease worker");
|
|
return await scope.ServiceProvider.GetRequiredService<IBackgroundJobService>()
|
|
.ProcessPendingAsync($"{workerId}:{index}", batchSize, includeImmediateJobs: true, cancellationToken: cancellationToken);
|
|
}
|
|
}
|
|
|
|
internal sealed class AuthorizationCacheInvalidationWorker(
|
|
IServiceScopeFactory scopeFactory,
|
|
IPeriodicProcessorLock processorLock,
|
|
WorkerStateReporter stateReporter,
|
|
IOptions<WorkerOptions> options,
|
|
ILogger<AuthorizationCacheInvalidationWorker> logger)
|
|
: PeriodicWorker(logger, processorLock, stateReporter, "authorization-cache-invalidations",
|
|
TimeSpan.FromSeconds(options.Value.JobPollSeconds), options.Value.Enabled)
|
|
{
|
|
protected override async Task<int> ProcessAsync(CancellationToken cancellationToken)
|
|
{
|
|
await using var scope = scopeFactory.CreateAsyncScope();
|
|
InitializeSystem(scope.ServiceProvider, "Authorization cache invalidation worker");
|
|
return await scope.ServiceProvider.GetRequiredService<IAuthorizationCacheInvalidationProcessor>()
|
|
.ProcessPendingAsync(cancellationToken: cancellationToken);
|
|
}
|
|
}
|
|
|
|
internal sealed class CommercialBillingWorker(
|
|
IServiceScopeFactory scopeFactory,
|
|
IPeriodicProcessorLock processorLock,
|
|
WorkerStateReporter stateReporter,
|
|
IOptions<WorkerOptions> options,
|
|
ILogger<CommercialBillingWorker> logger)
|
|
: PeriodicWorker(logger, processorLock, stateReporter, "commercial-billing",
|
|
TimeSpan.FromSeconds(60), options.Value.Enabled)
|
|
{
|
|
protected override async Task<int> ProcessAsync(CancellationToken cancellationToken)
|
|
{
|
|
await using var scope = scopeFactory.CreateAsyncScope();
|
|
InitializeSystem(scope.ServiceProvider, "Commercial renewal, receivable, dunning and refund worker");
|
|
return await scope.ServiceProvider.GetRequiredService<ICommercialBillingProcessor>()
|
|
.ProcessDueAsync(cancellationToken);
|
|
}
|
|
}
|