using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; using System.Text.Json; using Tiku.Application.Content; using Tiku.Application.Jobs; using Tiku.Application.Security; using Tiku.Application.Tenancy; using Tiku.Domain.Commerce; using Tiku.Domain.Common; using Tiku.Domain.Content; using Tiku.Domain.Operations; using Tiku.Infrastructure.Persistence; namespace Tiku.Infrastructure.Jobs; internal sealed class BackgroundJobService( TikuDbContext dbContext, ITenantExecutionScope tenantExecutionScope, IFeatureAccessService featureAccessService, IBackgroundJobDispatcher backgroundJobDispatcher) : IBackgroundJobService { private static readonly TimeSpan LeaseDuration = TimeSpan.FromMinutes(5); public async Task EnqueueAsync( CreateBackgroundJobCommand command, CancellationToken cancellationToken = default) { var normalizedJobType = NormalizeJobType(command.JobType); if (!(await featureAccessService.EvaluateAsync( command.TenantId, ResolveRequiredFeature(normalizedJobType, command.Payload), FeatureAccessOperation.Write, cancellationToken)).Allowed) { throw new InvalidOperationException("Tenant feature entitlement does not allow this background job."); } var quotaMetric = ResolveQuotaMetric(normalizedJobType); var quotaConsumed = false; if (quotaMetric is not null) { var quota = (await featureAccessService.GetQuotaSummaryAsync(command.TenantId, cancellationToken)) .SingleOrDefault(value => value.MetricCode == quotaMetric); if (quota is not null) { quotaConsumed = await featureAccessService.TryConsumeQuotaAsync( command.TenantId, quotaMetric, 1, cancellationToken); if (!quotaConsumed) { throw new FeatureAccessException("The background job quota has been exhausted.", "feature_quota_exhausted"); } } } var job = new BackgroundJob { TenantId = command.TenantId, JobType = normalizedJobType, Payload = command.Payload, RunAfter = command.RunAfter, MaxRetries = Math.Clamp(command.MaxRetries, 0, 20) }; dbContext.BackgroundJobs.Add(job); try { await dbContext.SaveChangesAsync(cancellationToken); } catch { if (quotaConsumed && quotaMetric is not null) { await featureAccessService.ReleaseQuotaAsync(command.TenantId, quotaMetric, 1, CancellationToken.None); } throw; } if (job.RunAfter is null && backgroundJobDispatcher.IsEnabled) { await backgroundJobDispatcher.DispatchAsync( job.Id, job.TenantId, job.JobType, job.Id.ToString("N"), cancellationToken); } return ToItem(job); } private static string? ResolveQuotaMetric(string jobType) => jobType switch { "content_import" => SaasQuotaMetricCatalog.ImportCount, "content_export" => SaasQuotaMetricCatalog.ExportCount, _ => null }; public async Task ProcessPendingAsync( string workerId, int batchSize, bool includeImmediateJobs = true, CancellationToken cancellationToken = default) { var now = DateTimeOffset.UtcNow; var leaseExpiresAt = now.Add(LeaseDuration); var claimedIds = await dbContext.Database.SqlQuery($""" UPDATE background_jobs AS job SET status = 'processing', locked_by = {workerId}, lock_expires_at = {leaseExpiresAt}, started_at = COALESCE(started_at, {now}), updated_at = {now} WHERE job.id IN ( SELECT candidate.id FROM background_jobs AS candidate WHERE ( (candidate.status = 'pending' AND ({includeImmediateJobs} OR candidate.run_after IS NOT NULL) AND (candidate.run_after IS NULL OR candidate.run_after <= {now})) OR (candidate.status = 'processing' AND candidate.lock_expires_at <= {now}) ) ORDER BY candidate.created_at, candidate.id FOR UPDATE SKIP LOCKED LIMIT {Math.Clamp(batchSize, 1, 100)} ) RETURNING job.id AS "Value" """) .ToArrayAsync(cancellationToken); var processed = 0; dbContext.ChangeTracker.Clear(); foreach (var jobId in claimedIds) { cancellationToken.ThrowIfCancellationRequested(); var job = await dbContext.BackgroundJobs.SingleAsync(value => value.Id == jobId, cancellationToken); if (await ProcessJobAsync(job, workerId, alreadyClaimed: true, cancellationToken)) processed++; dbContext.ChangeTracker.Clear(); } return processed; } public async Task ProcessRequestedAsync( Guid jobId, Guid tenantId, string jobType, string workerId, CancellationToken cancellationToken = default) { var normalizedJobType = NormalizeJobType(jobType); var claimed = await dbContext.BackgroundJobs .Where(item => item.Id == jobId && item.TenantId == tenantId && item.JobType == normalizedJobType && item.Status == BackgroundJobStatus.Pending && item.RunAfter == null) .ExecuteUpdateAsync(setters => setters .SetProperty(item => item.Status, BackgroundJobStatus.Processing) .SetProperty(item => item.LockedBy, workerId) .SetProperty(item => item.LockExpiresAt, DateTimeOffset.UtcNow.Add(LeaseDuration)) .SetProperty(item => item.StartedAt, DateTimeOffset.UtcNow), cancellationToken); if (claimed == 0) { return false; } dbContext.ChangeTracker.Clear(); var job = await dbContext.BackgroundJobs.SingleOrDefaultAsync( item => item.Id == jobId && item.TenantId == tenantId, cancellationToken); if (job is null) { throw new InvalidOperationException("The requested background job does not exist in the target tenant."); } if (!string.Equals(job.JobType, normalizedJobType, StringComparison.Ordinal)) { throw new InvalidOperationException("The requested background job type does not match the persisted job."); } return await ProcessJobAsync(job, workerId, alreadyClaimed: true, cancellationToken); } private async Task ProcessJobAsync( BackgroundJob job, string workerId, bool alreadyClaimed, CancellationToken cancellationToken) { if ((!alreadyClaimed && job.Status != BackgroundJobStatus.Pending) || (alreadyClaimed && (job.Status != BackgroundJobStatus.Processing || job.LockedBy != workerId))) { return false; } if (!(await featureAccessService.EvaluateAsync( job.TenantId, ResolveRequiredFeature(job.JobType, job.Payload), FeatureAccessOperation.Write, cancellationToken)).Allowed) { await CompleteAsync( job, workerId, BackgroundJobStatus.Failed, JsonDefaults.Object(), "Tenant feature entitlement was revoked before job execution.", cancellationToken); return true; } if (!alreadyClaimed) { var now = DateTimeOffset.UtcNow; job.Status = BackgroundJobStatus.Processing; job.LockedBy = workerId; job.LockExpiresAt = now.Add(LeaseDuration); job.StartedAt = now; await dbContext.SaveChangesAsync(cancellationToken); } try { var result = await tenantExecutionScope.ExecuteAsync( new SystemScopeRequest( job.TenantId, SystemScopeCallerType.Worker, workerId, $"Background job {job.JobType}", job.Id.ToString("N")), (provider, token) => ProcessCoreAsync(provider, job, token), cancellationToken); job.Status = BackgroundJobStatus.Succeeded; job.CompletedAt = DateTimeOffset.UtcNow; job.LastError = null; job.Result = result; } 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 = job.Status == BackgroundJobStatus.Pending ? DateTimeOffset.UtcNow.AddSeconds(Math.Min(300, 10 * job.RetryCount)) : null; } finally { await CompleteAsync(job, workerId, job.Status, job.Result, job.LastError, cancellationToken); } return true; } private async Task CompleteAsync( BackgroundJob job, string workerId, BackgroundJobStatus status, JsonElement result, string? lastError, CancellationToken cancellationToken) { await dbContext.BackgroundJobs .Where(value => value.Id == job.Id && value.LockedBy == workerId) .ExecuteUpdateAsync(setters => setters .SetProperty(value => value.Status, status) .SetProperty(value => value.RetryCount, job.RetryCount) .SetProperty(value => value.RunAfter, job.RunAfter) .SetProperty(value => value.CompletedAt, job.CompletedAt) .SetProperty(value => value.LastError, lastError) .SetProperty(value => value.OutputAssetId, job.OutputAssetId) .SetProperty(value => value.Result, result) .SetProperty(value => value.LockedBy, (string?)null) .SetProperty(value => value.LockExpiresAt, (DateTimeOffset?)null), cancellationToken); } 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 async Task ProcessCoreAsync( IServiceProvider scopedProvider, BackgroundJob job, CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); var scopedDbContext = scopedProvider.GetRequiredService(); return job.JobType switch { "content_export" => await ProcessContentExportAsync(scopedDbContext, job, cancellationToken), "content_import" => await ProcessContentImportAsync(scopedProvider, job, cancellationToken), "asset_security_scan" => throw new NotSupportedException("asset_security_scan requires a configured scanner provider before it can write scan results."), "statistics_aggregation" => await ProcessStatisticsAggregationAsync(scopedProvider, scopedDbContext, job, cancellationToken), "commerce_reconciliation" => await ProcessCommerceReconciliationAsync(scopedDbContext, job, cancellationToken), "tenant_domain_recheck" => await ProcessTenantDomainRecheckAsync(scopedProvider, cancellationToken), _ => throw new InvalidOperationException($"Unsupported background job type '{job.JobType}'.") }; } private static async Task ProcessContentImportAsync( IServiceProvider scopedProvider, BackgroundJob job, CancellationToken cancellationToken) { var directContentService = scopedProvider.GetRequiredService(); var createdBy = GetJsonGuid(job.Payload, "createdBy") ?? Guid.Empty; if (createdBy == Guid.Empty) { throw new InvalidOperationException("content_import job requires createdBy."); } var command = new SimpleImportCommand( GetJsonString(job.Payload, "importType") ?? throw new InvalidOperationException("content_import job requires importType."), GetJsonString(job.Payload, "sourceFormat"), GetJsonString(job.Payload, "sourceName"), GetJsonGuid(job.Payload, "regionId"), GetJsonGuid(job.Payload, "entryId"), GetJsonGuid(job.Payload, "contentNodeId"), GetJsonGuid(job.Payload, "subjectId"), GetJsonGuid(job.Payload, "categoryId"), GetJsonGuid(job.Payload, "questionBankId"), GetJsonGuid(job.Payload, "collectionId"), GetJsonArray(job.Payload, "items"), false); var result = await directContentService.ExecuteImportAsync( new DirectContentActor(job.TenantId, createdBy), command, cancellationToken); return JsonSerializer.SerializeToElement(new { importJobId = result.Job.Id, result.Job.ImportType, result.Job.Status, result.Job.TotalCount, result.Job.InsertedCount, result.Job.UpdatedCount, result.Job.ErrorCount }); } private async Task ProcessContentExportAsync( TikuDbContext scopedDbContext, BackgroundJob job, CancellationToken cancellationToken) { var exportType = GetJsonString(job.Payload, "exportType") ?? "summary"; var assetKey = $"background-jobs/{job.Id:N}/content-export.json"; var asset = await scopedDbContext.ContentAssets.SingleOrDefaultAsync( item => item.TenantId == job.TenantId && item.AssetKey == assetKey, cancellationToken); if (asset is null) { asset = new ContentAsset { TenantId = job.TenantId, AssetKey = assetKey, AssetType = ContentAssetType.Document, StorageProvider = AssetStorageProvider.ExternalUrl, UploadStatus = AssetUploadStatus.Verified, SecurityScanStatus = AssetSecurityScanStatus.NotRequired, Source = "background_job" }; scopedDbContext.ContentAssets.Add(asset); } var questionBankCount = await scopedDbContext.QuestionBanks.CountAsync(item => item.TenantId == job.TenantId, cancellationToken); var questionCount = await scopedDbContext.Questions.CountAsync(item => item.TenantId == job.TenantId, cancellationToken); var studentCount = await scopedDbContext.StudentProfiles.CountAsync(item => item.TenantId == job.TenantId, cancellationToken); asset.FileName = $"content-export-{DateTimeOffset.UtcNow:yyyyMMddHHmmss}.json"; asset.Title = "Content export manifest"; asset.Description = $"Generated content export manifest for {exportType}."; asset.ObjectKey = assetKey; asset.MimeType = "application/json"; asset.Metadata = JsonSerializer.SerializeToElement(new { exportType, generatedAt = DateTimeOffset.UtcNow, questionBankCount, questionCount, studentCount, payload = job.Payload }); await scopedDbContext.SaveChangesAsync(cancellationToken); job.OutputAssetId = asset.Id; return JsonSerializer.SerializeToElement(new { outputAssetId = asset.Id, asset.AssetKey, questionBankCount, questionCount, studentCount }); } private async Task ProcessCommerceReconciliationAsync( TikuDbContext scopedDbContext, BackgroundJob job, CancellationToken cancellationToken) { var provider = NormalizeProvider(GetJsonString(job.Payload, "provider")); var hasProviderConfig = await scopedDbContext.TenantExternalProviders.AnyAsync( item => item.TenantId == job.TenantId && item.Capability == Tiku.Domain.Tenancy.TenantExternalProviderCapability.Payment && item.Provider == provider && item.Status == Tiku.Domain.Tenancy.TenantExternalProviderStatus.Active, cancellationToken); if (!hasProviderConfig) { throw new InvalidOperationException($"Active payment provider '{provider}' is required for commerce reconciliation job."); } var billDate = GetJsonDateOnly(job.Payload, "billDate") ?? DateOnly.FromDateTime(DateTime.UtcNow.Date); var billType = GetJsonEnum(job.Payload, "billType", ReconciliationBillType.Combined); var sourceHash = $"background-job:{job.Id:N}"; var batch = await scopedDbContext.CommerceReconciliationBatches.SingleOrDefaultAsync( item => item.TenantId == job.TenantId && item.Provider == provider && item.Source == ReconciliationSource.ProviderDownload && item.SourceHash == sourceHash, cancellationToken); if (batch is null) { batch = new CommerceReconciliationBatch { TenantId = job.TenantId, Provider = provider, BillDate = billDate, BillType = billType, Source = ReconciliationSource.ProviderDownload, SourceName = $"provider-bill:{provider}:{billDate:yyyyMMdd}", SourceHash = sourceHash, Status = ReconciliationBatchStatus.Pending, Metadata = JsonSerializer.SerializeToElement(new { jobId = job.Id, note = "Provider bill job created the reconciliation batch; provider download/parser is handled by a dedicated provider processor." }) }; scopedDbContext.CommerceReconciliationBatches.Add(batch); await scopedDbContext.SaveChangesAsync(cancellationToken); } return JsonSerializer.SerializeToElement(new { batchId = batch.Id, provider, billDate, billType = billType.ToString(), status = batch.Status.ToString() }); } private static async Task ProcessTenantDomainRecheckAsync( IServiceProvider scopedProvider, CancellationToken cancellationToken) { var lifecycleService = scopedProvider.GetRequiredService(); var processed = await lifecycleService.ProcessPendingAsync(cancellationToken); return JsonSerializer.SerializeToElement(new { processed }); } private async Task ProcessStatisticsAggregationAsync( IServiceProvider scopedProvider, TikuDbContext scopedDbContext, BackgroundJob job, CancellationToken cancellationToken) { var since = DateTimeOffset.UtcNow.AddDays(-7); var activeLearnerCount = await scopedDbContext.PracticeSessions .Where(item => item.TenantId == job.TenantId && item.StartedAt >= since) .Select(item => item.UserId) .Distinct() .CountAsync(cancellationToken); var paidOrderCount = await scopedDbContext.Orders .CountAsync(item => item.TenantId == job.TenantId && item.Status == OrderStatus.Paid, cancellationToken); var revenueCents = await scopedDbContext.Orders .Where(item => item.TenantId == job.TenantId && (item.Status == OrderStatus.Paid || item.Status == OrderStatus.PartiallyRefunded || item.Status == OrderStatus.Refunded)) .SumAsync(item => item.AmountCents - item.RefundedAmountCents, cancellationToken); var quotaUsage = await scopedProvider.GetRequiredService() .ReconcileTenantAsync( new ReconcileFeatureUsageRequest( job.TenantId, SystemScopeCallerType.Worker, nameof(BackgroundJobService), "Reconcile tenant current feature usage during statistics aggregation", $"feature-usage-{job.Id:N}"), cancellationToken); return JsonSerializer.SerializeToElement(new { since, activeLearnerCount, paidOrderCount, revenueCents, quotaUsage }); } private static string NormalizeJobType(string jobType) { return jobType.Trim().ToLowerInvariant(); } private static string ResolveRequiredFeature(string jobType, JsonElement payload) => jobType switch { "content_import" => SaasFeatureCatalog.ResolveContentImportFeature(GetJsonString(payload, "importType")) ?? throw new InvalidOperationException("Background content import type is not supported."), "content_export" or "asset_security_scan" => SaasFeatureCatalog.PrivateQuestionBank, "commerce_reconciliation" => SaasFeatureCatalog.StudentStore, "statistics_aggregation" or "tenant_domain_recheck" => SaasFeatureCatalog.CoreBackoffice, _ => SaasFeatureCatalog.CoreBackoffice }; private static string NormalizeProvider(string? provider) { var normalized = (provider ?? string.Empty).Trim().ToLowerInvariant(); return string.IsNullOrWhiteSpace(normalized) ? throw new InvalidOperationException("Background job provider is required.") : normalized; } private static string? GetJsonString(JsonElement element, string propertyName) { return element.ValueKind == JsonValueKind.Object && element.TryGetProperty(propertyName, out var property) && property.ValueKind == JsonValueKind.String ? property.GetString() : null; } private static Guid? GetJsonGuid(JsonElement element, string propertyName) { if (element.ValueKind != JsonValueKind.Object || !element.TryGetProperty(propertyName, out var property)) { return null; } return property.ValueKind == JsonValueKind.String && Guid.TryParse(property.GetString(), out var value) ? value : null; } private static IReadOnlyCollection GetJsonArray(JsonElement element, string propertyName) { if (element.ValueKind != JsonValueKind.Object || !element.TryGetProperty(propertyName, out var property) || property.ValueKind != JsonValueKind.Array) { return []; } return property.EnumerateArray().Select(item => item.Clone()).ToArray(); } private static DateOnly? GetJsonDateOnly(JsonElement element, string propertyName) { var value = GetJsonString(element, propertyName); return DateOnly.TryParse(value, out var parsed) ? parsed : null; } private static TEnum GetJsonEnum(JsonElement element, string propertyName, TEnum fallback) where TEnum : struct { var value = GetJsonString(element, propertyName); return Enum.TryParse(value, true, out var parsed) ? parsed : fallback; } 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); } }