using Microsoft.EntityFrameworkCore; using Tiku.Application.Jobs; using Tiku.Application.Security; using Tiku.Domain.Common; using Tiku.Domain.Operations; using Tiku.Infrastructure.Persistence; namespace Tiku.Infrastructure.Jobs; internal sealed class BackgroundJobService( TikuDbContext dbContext, ITenantExecutionScope tenantExecutionScope) : IBackgroundJobService { private static readonly TimeSpan LeaseDuration = TimeSpan.FromMinutes(5); public async Task EnqueueAsync( CreateBackgroundJobCommand command, CancellationToken cancellationToken = default) { var job = new BackgroundJob { TenantId = command.TenantId, JobType = NormalizeJobType(command.JobType), Payload = command.Payload, RunAfter = command.RunAfter, MaxRetries = Math.Clamp(command.MaxRetries, 0, 20) }; dbContext.BackgroundJobs.Add(job); await dbContext.SaveChangesAsync(cancellationToken); return ToItem(job); } public async Task ProcessPendingAsync( string workerId, int batchSize, CancellationToken cancellationToken = default) { var now = DateTimeOffset.UtcNow; var jobs = await dbContext.BackgroundJobs .Where(job => job.Status == BackgroundJobStatus.Pending && (job.RunAfter == null || job.RunAfter <= now)) .OrderBy(job => job.CreatedAt) .Take(Math.Clamp(batchSize, 1, 100)) .ToArrayAsync(cancellationToken); var processed = 0; foreach (var job in jobs) { cancellationToken.ThrowIfCancellationRequested(); job.Status = BackgroundJobStatus.Processing; job.LockedBy = workerId; job.LockExpiresAt = now.Add(LeaseDuration); job.StartedAt = now; await dbContext.SaveChangesAsync(cancellationToken); try { await tenantExecutionScope.ExecuteAsync( job.TenantId, $"Background job {job.JobType}", (_, token) => ProcessCoreAsync(job, token), cancellationToken); job.Status = BackgroundJobStatus.Succeeded; job.CompletedAt = DateTimeOffset.UtcNow; job.LastError = null; job.Result = JsonDefaults.Object(); } catch (Exception exception) when (exception is not OperationCanceledException) { job.RetryCount++; job.LastError = exception.Message; job.Status = job.RetryCount > job.MaxRetries ? BackgroundJobStatus.Failed : BackgroundJobStatus.Pending; job.RunAfter = DateTimeOffset.UtcNow.AddSeconds(Math.Min(300, 10 * job.RetryCount)); } finally { job.LockedBy = null; job.LockExpiresAt = null; processed++; await dbContext.SaveChangesAsync(cancellationToken); } } return processed; } public async Task> ListAsync( Guid tenantId, string? jobType = null, int limit = 50, CancellationToken cancellationToken = default) { var query = dbContext.BackgroundJobs.AsNoTracking() .Where(job => job.TenantId == tenantId); if (!string.IsNullOrWhiteSpace(jobType)) { var normalized = NormalizeJobType(jobType); query = query.Where(job => job.JobType == normalized); } var jobs = await query .OrderByDescending(job => job.CreatedAt) .Take(Math.Clamp(limit, 1, 200)) .ToArrayAsync(cancellationToken); return jobs.Select(ToItem).ToArray(); } private static Task ProcessCoreAsync(BackgroundJob job, CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); return job.JobType switch { "content_export" => Task.CompletedTask, "content_import" => Task.CompletedTask, "asset_security_scan" => Task.CompletedTask, "statistics_aggregation" => Task.CompletedTask, "commerce_reconciliation" => Task.CompletedTask, "tenant_domain_recheck" => Task.CompletedTask, _ => throw new InvalidOperationException($"Unsupported background job type '{job.JobType}'.") }; } private static string NormalizeJobType(string jobType) { return jobType.Trim().ToLowerInvariant(); } private static BackgroundJobItem ToItem(BackgroundJob job) { return new BackgroundJobItem( job.Id, job.TenantId, job.JobType, job.Status, job.RetryCount, job.MaxRetries, job.RunAfter, job.StartedAt, job.CompletedAt, job.LastError, job.OutputAssetId, job.Result); } }