diff --git a/Tiku.Application/Jobs/BackgroundJobModels.cs b/Tiku.Application/Jobs/BackgroundJobModels.cs index 43ff1b9..6d65ebc 100644 --- a/Tiku.Application/Jobs/BackgroundJobModels.cs +++ b/Tiku.Application/Jobs/BackgroundJobModels.cs @@ -103,11 +103,29 @@ public sealed record BackgroundJobExecutionContext( string JobType, JsonElement Payload); +public sealed record BackgroundJobHandlerResult( + JsonElement Result, + Guid? OutputAssetId = null); + +public sealed class BackgroundJobHandlerException(string code, string message, Exception? innerException = null) + : Exception(message, innerException) +{ + public string Code { get; } = code; +} + public interface IBackgroundJobHandler { string JobType { get; } - Task HandleAsync( + Task HandleAsync( BackgroundJobExecutionContext context, CancellationToken cancellationToken = default); -} \ No newline at end of file + + Task HandleFailureAsync( + BackgroundJobExecutionContext context, + Exception exception, + CancellationToken cancellationToken = default) + { + return Task.CompletedTask; + } +} diff --git a/Tiku.Infrastructure/Assets/Security/AssetSecurityScanJobHandler.cs b/Tiku.Infrastructure/Assets/Security/AssetSecurityScanJobHandler.cs new file mode 100644 index 0000000..3354a4d --- /dev/null +++ b/Tiku.Infrastructure/Assets/Security/AssetSecurityScanJobHandler.cs @@ -0,0 +1,118 @@ +using System.Text.Json; +using Microsoft.EntityFrameworkCore; +using Tiku.Application.Assets; +using Tiku.Application.Jobs; +using Tiku.Application.Storage; +using Tiku.Domain.Content; +using Tiku.Infrastructure.Jobs; +using Tiku.Infrastructure.Persistence; + +namespace Tiku.Infrastructure.Assets.Security; + +internal sealed class AssetSecurityScanJobHandler( + TikuDbContext dbContext, + IAssetSecurityScanner scanner, + IObjectStorageService storage) : IBackgroundJobHandler +{ + public string JobType => "asset_security_scan"; + + public async Task HandleAsync( + BackgroundJobExecutionContext context, + CancellationToken cancellationToken = default) + { + var assetId = BackgroundJobPayload.GetGuid(context.Payload, "assetId") ?? + throw new InvalidOperationException("asset_security_scan job requires assetId."); + var asset = await dbContext.ContentAssets.SingleOrDefaultAsync( + item => item.TenantId == context.TenantId && item.Id == assetId, + cancellationToken) ?? throw new InvalidOperationException("Asset security scan target was not found."); + if (asset.UploadStatus != AssetUploadStatus.Verified || + string.IsNullOrWhiteSpace(asset.Bucket) || + string.IsNullOrWhiteSpace(asset.ObjectKey)) + throw new InvalidOperationException("Asset must have a verified object location before security scanning."); + + asset.SecurityScanStatus = AssetSecurityScanStatus.Scanning; + await dbContext.SaveChangesAsync(cancellationToken); + try + { + await using var content = await storage.OpenReadAsync( + new ObjectStorageReadRequest( + context.TenantId, + asset.StorageProvider switch + { + AssetStorageProvider.AliyunOss => ObjectStorageProviders.AliyunOss, + AssetStorageProvider.LocalDev => ObjectStorageProviders.LocalDev, + _ => throw new InvalidOperationException( + "Asset storage provider does not support security scanning.") + }, + asset.Bucket, + asset.ObjectKey), + cancellationToken); + var result = + await scanner.ScanAsync(content, asset.VerifiedSizeBytes ?? asset.FileSizeBytes, cancellationToken); + var infected = result.Verdict == AssetSecurityScanVerdict.Infected; + asset.SecurityScanStatus = infected ? AssetSecurityScanStatus.Failed : AssetSecurityScanStatus.Passed; + asset.SecurityScannedAt = DateTimeOffset.UtcNow; + asset.SecurityScanProvider = result.Provider; + asset.SecurityScanSummary = JsonSerializer.SerializeToElement(new + { + verdict = result.Verdict.ToString(), + result.Signature, + result.BytesScanned + }); + dbContext.ContentAssetSecurityScanEvents.Add(new ContentAssetSecurityScanEvent + { + TenantId = context.TenantId, + AssetId = asset.Id, + Provider = result.Provider, + ScanStatus = asset.SecurityScanStatus, + RiskLevel = infected ? AssetSecurityRiskLevel.Critical : AssetSecurityRiskLevel.None, + IssueCodes = infected ? [result.Signature ?? "malware_detected"] : [], + Details = asset.SecurityScanSummary + }); + await dbContext.SaveChangesAsync(cancellationToken); + return new BackgroundJobHandlerResult(JsonSerializer.SerializeToElement(new + { + assetId = asset.Id, + status = asset.SecurityScanStatus.ToString(), + result.Signature, + result.BytesScanned + })); + } + catch (AssetSecurityScannerException exception) + { + throw new BackgroundJobHandlerException(exception.Code, exception.Message, exception); + } + } + + public async Task HandleFailureAsync( + BackgroundJobExecutionContext context, + Exception exception, + CancellationToken cancellationToken = default) + { + if (exception is not BackgroundJobHandlerException handlerException || + handlerException.InnerException is not AssetSecurityScannerException) + return; + + var assetId = BackgroundJobPayload.GetGuid(context.Payload, "assetId"); + if (assetId is null) return; + var asset = await dbContext.ContentAssets.SingleOrDefaultAsync( + item => item.TenantId == context.TenantId && item.Id == assetId, + cancellationToken); + if (asset is null) return; + + asset.SecurityScanStatus = AssetSecurityScanStatus.Pending; + asset.SecurityScanProvider = "clamav"; + asset.SecurityScanSummary = JsonSerializer.SerializeToElement(new { errorCode = handlerException.Code }); + dbContext.ContentAssetSecurityScanEvents.Add(new ContentAssetSecurityScanEvent + { + TenantId = context.TenantId, + AssetId = asset.Id, + Provider = "clamav", + ScanStatus = AssetSecurityScanStatus.Pending, + RiskLevel = AssetSecurityRiskLevel.None, + IssueCodes = [handlerException.Code], + Details = asset.SecurityScanSummary + }); + await dbContext.SaveChangesAsync(cancellationToken); + } +} diff --git a/Tiku.Infrastructure/Commerce/Reconciliation/CommerceReconciliationJobHandler.cs b/Tiku.Infrastructure/Commerce/Reconciliation/CommerceReconciliationJobHandler.cs new file mode 100644 index 0000000..8d3e977 --- /dev/null +++ b/Tiku.Infrastructure/Commerce/Reconciliation/CommerceReconciliationJobHandler.cs @@ -0,0 +1,82 @@ +using System.Text.Json; +using Microsoft.EntityFrameworkCore; +using Tiku.Application.Jobs; +using Tiku.Domain.Commerce; +using Tiku.Domain.Tenancy; +using Tiku.Infrastructure.Jobs; +using Tiku.Infrastructure.Persistence; + +namespace Tiku.Infrastructure.Commerce.Reconciliation; + +internal sealed class CommerceReconciliationJobHandler(TikuDbContext dbContext) : IBackgroundJobHandler +{ + public string JobType => "commerce_reconciliation"; + + public async Task HandleAsync( + BackgroundJobExecutionContext context, + CancellationToken cancellationToken = default) + { + var provider = NormalizeProvider(BackgroundJobPayload.GetString(context.Payload, "provider")); + var hasProviderConfig = await dbContext.TenantExternalProviders.AnyAsync( + item => + item.TenantId == context.TenantId && + item.Capability == TenantExternalProviderCapability.Payment && + item.Provider == provider && + item.Status == TenantExternalProviderStatus.Active, + cancellationToken); + if (!hasProviderConfig) + throw new InvalidOperationException( + $"Active payment provider '{provider}' is required for commerce reconciliation job."); + + var billDate = BackgroundJobPayload.GetDateOnly(context.Payload, "billDate") ?? + DateOnly.FromDateTime(DateTime.UtcNow.Date); + var billType = BackgroundJobPayload.GetEnum(context.Payload, "billType", ReconciliationBillType.Combined); + var sourceHash = $"background-job:{context.JobId:N}"; + var batch = await dbContext.CommerceReconciliationBatches.SingleOrDefaultAsync( + item => + item.TenantId == context.TenantId && + item.Provider == provider && + item.Source == ReconciliationSource.ProviderDownload && + item.SourceHash == sourceHash, + cancellationToken); + if (batch is null) + { + batch = new CommerceReconciliationBatch + { + TenantId = context.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 = context.JobId, + note = + "Provider bill job created the reconciliation batch; provider download/parser is handled by a dedicated provider processor." + }) + }; + dbContext.CommerceReconciliationBatches.Add(batch); + await dbContext.SaveChangesAsync(cancellationToken); + } + + return new BackgroundJobHandlerResult(JsonSerializer.SerializeToElement(new + { + batchId = batch.Id, + provider, + billDate, + billType = billType.ToString(), + status = batch.Status.ToString() + })); + } + + 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; + } +} diff --git a/Tiku.Infrastructure/Content/Exports/ContentExportJobHandler.cs b/Tiku.Infrastructure/Content/Exports/ContentExportJobHandler.cs new file mode 100644 index 0000000..4d16cb9 --- /dev/null +++ b/Tiku.Infrastructure/Content/Exports/ContentExportJobHandler.cs @@ -0,0 +1,70 @@ +using System.Text.Json; +using Microsoft.EntityFrameworkCore; +using Tiku.Application.Jobs; +using Tiku.Domain.Content; +using Tiku.Infrastructure.Jobs; +using Tiku.Infrastructure.Persistence; + +namespace Tiku.Infrastructure.Content.Exports; + +internal sealed class ContentExportJobHandler(TikuDbContext dbContext) : IBackgroundJobHandler +{ + public string JobType => "content_export"; + + public async Task HandleAsync( + BackgroundJobExecutionContext context, + CancellationToken cancellationToken = default) + { + var exportType = BackgroundJobPayload.GetString(context.Payload, "exportType") ?? "summary"; + var assetKey = $"background-jobs/{context.JobId:N}/content-export.json"; + var asset = await dbContext.ContentAssets.SingleOrDefaultAsync( + item => item.TenantId == context.TenantId && item.AssetKey == assetKey, + cancellationToken); + if (asset is null) + { + asset = new ContentAsset + { + TenantId = context.TenantId, + AssetKey = assetKey, + AssetType = ContentAssetType.Document, + StorageProvider = AssetStorageProvider.ExternalUrl, + UploadStatus = AssetUploadStatus.Verified, + SecurityScanStatus = AssetSecurityScanStatus.NotRequired, + Source = "background_job" + }; + dbContext.ContentAssets.Add(asset); + } + + var questionBankCount = + await dbContext.QuestionBanks.CountAsync(item => item.TenantId == context.TenantId, cancellationToken); + var questionCount = + await dbContext.Questions.CountAsync(item => item.TenantId == context.TenantId, cancellationToken); + var studentCount = + await dbContext.StudentProfiles.CountAsync(item => item.TenantId == context.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 = context.Payload + }); + await dbContext.SaveChangesAsync(cancellationToken); + return new BackgroundJobHandlerResult( + JsonSerializer.SerializeToElement(new + { + outputAssetId = asset.Id, + asset.AssetKey, + questionBankCount, + questionCount, + studentCount + }), + asset.Id); + } +} diff --git a/Tiku.Infrastructure/Content/Imports/ContentImportJobHandler.cs b/Tiku.Infrastructure/Content/Imports/ContentImportJobHandler.cs new file mode 100644 index 0000000..b79d442 --- /dev/null +++ b/Tiku.Infrastructure/Content/Imports/ContentImportJobHandler.cs @@ -0,0 +1,48 @@ +using System.Text.Json; +using Tiku.Application.Content; +using Tiku.Application.Jobs; +using Tiku.Infrastructure.Jobs; + +namespace Tiku.Infrastructure.Content.Imports; + +internal sealed class ContentImportJobHandler(IDirectContentService directContentService) : IBackgroundJobHandler +{ + public string JobType => "content_import"; + + public async Task HandleAsync( + BackgroundJobExecutionContext context, + CancellationToken cancellationToken = default) + { + var createdBy = BackgroundJobPayload.GetGuid(context.Payload, "createdBy") ?? Guid.Empty; + if (createdBy == Guid.Empty) throw new InvalidOperationException("content_import job requires createdBy."); + + var command = new SimpleImportCommand( + BackgroundJobPayload.GetString(context.Payload, "importType") ?? + throw new InvalidOperationException("content_import job requires importType."), + BackgroundJobPayload.GetString(context.Payload, "sourceFormat"), + BackgroundJobPayload.GetString(context.Payload, "sourceName"), + BackgroundJobPayload.GetGuid(context.Payload, "regionId"), + BackgroundJobPayload.GetGuid(context.Payload, "entryId"), + BackgroundJobPayload.GetGuid(context.Payload, "contentNodeId"), + BackgroundJobPayload.GetGuid(context.Payload, "subjectId"), + BackgroundJobPayload.GetGuid(context.Payload, "categoryId"), + BackgroundJobPayload.GetGuid(context.Payload, "questionBankId"), + BackgroundJobPayload.GetGuid(context.Payload, "collectionId"), + BackgroundJobPayload.GetArray(context.Payload, "items"), + false); + var result = await directContentService.ExecuteImportAsync( + new DirectContentActor(context.TenantId, createdBy), + command, + cancellationToken); + return new BackgroundJobHandlerResult(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 + })); + } +} diff --git a/Tiku.Infrastructure/Jobs/BackgroundJobService.cs b/Tiku.Infrastructure/Jobs/BackgroundJobService.cs index 0d88168..94cafca 100644 --- a/Tiku.Infrastructure/Jobs/BackgroundJobService.cs +++ b/Tiku.Infrastructure/Jobs/BackgroundJobService.cs @@ -10,4 +10,4 @@ internal sealed partial class BackgroundJobService( IFeatureAccessService featureAccessService) : IBackgroundJobService { private static readonly TimeSpan LeaseDuration = TimeSpan.FromMinutes(5); -} \ No newline at end of file +} diff --git a/Tiku.Infrastructure/Jobs/Foundation/BackgroundJobService.Foundation.cs b/Tiku.Infrastructure/Jobs/Foundation/BackgroundJobService.Foundation.cs index e1c2fe3..698f2f4 100644 --- a/Tiku.Infrastructure/Jobs/Foundation/BackgroundJobService.Foundation.cs +++ b/Tiku.Infrastructure/Jobs/Foundation/BackgroundJobService.Foundation.cs @@ -36,14 +36,6 @@ internal sealed partial class BackgroundJobService }; } - 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 && @@ -53,40 +45,6 @@ internal sealed partial class BackgroundJobService : 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( @@ -107,4 +65,4 @@ internal sealed partial class BackgroundJobService job.OutputAssetId, job.Result); } -} \ No newline at end of file +} diff --git a/Tiku.Infrastructure/Jobs/Handlers/BackgroundJobHandlerRegistry.cs b/Tiku.Infrastructure/Jobs/Handlers/BackgroundJobHandlerRegistry.cs new file mode 100644 index 0000000..ba399ed --- /dev/null +++ b/Tiku.Infrastructure/Jobs/Handlers/BackgroundJobHandlerRegistry.cs @@ -0,0 +1,29 @@ +using Tiku.Application.Jobs; + +namespace Tiku.Infrastructure.Jobs; + +internal sealed class BackgroundJobHandlerRegistry +{ + private readonly IReadOnlyDictionary handlers; + + public BackgroundJobHandlerRegistry(IEnumerable handlers) + { + var registered = new Dictionary(StringComparer.Ordinal); + foreach (var handler in handlers) + { + var jobType = handler.JobType.Trim().ToLowerInvariant(); + if (!registered.TryAdd(jobType, handler)) + throw new InvalidOperationException($"Duplicate background job handler for '{jobType}'."); + } + + this.handlers = registered; + } + + public IBackgroundJobHandler GetRequired(string jobType) + { + var normalized = jobType.Trim().ToLowerInvariant(); + return handlers.TryGetValue(normalized, out var handler) + ? handler + : throw new InvalidOperationException($"Unsupported background job type '{normalized}'."); + } +} diff --git a/Tiku.Infrastructure/Jobs/Handlers/BackgroundJobPayload.cs b/Tiku.Infrastructure/Jobs/Handlers/BackgroundJobPayload.cs new file mode 100644 index 0000000..5aeb3a1 --- /dev/null +++ b/Tiku.Infrastructure/Jobs/Handlers/BackgroundJobPayload.cs @@ -0,0 +1,49 @@ +using System.Text.Json; + +namespace Tiku.Infrastructure.Jobs; + +internal static class BackgroundJobPayload +{ + public static string? GetString(JsonElement element, string propertyName) + { + return element.ValueKind == JsonValueKind.Object && + element.TryGetProperty(propertyName, out var property) && + property.ValueKind == JsonValueKind.String + ? property.GetString() + : null; + } + + public static Guid? GetGuid(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; + } + + public static IReadOnlyCollection GetArray(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(); + } + + public static DateOnly? GetDateOnly(JsonElement element, string propertyName) + { + var value = GetString(element, propertyName); + return DateOnly.TryParse(value, out var parsed) ? parsed : null; + } + + public static TEnum GetEnum(JsonElement element, string propertyName, TEnum fallback) + where TEnum : struct + { + var value = GetString(element, propertyName); + return Enum.TryParse(value, true, out var parsed) ? parsed : fallback; + } +} diff --git a/Tiku.Infrastructure/Jobs/Handlers/BackgroundJobService.Handlers.cs b/Tiku.Infrastructure/Jobs/Handlers/BackgroundJobService.Handlers.cs deleted file mode 100644 index cb567a5..0000000 --- a/Tiku.Infrastructure/Jobs/Handlers/BackgroundJobService.Handlers.cs +++ /dev/null @@ -1,567 +0,0 @@ -using System.Formats.Tar; -using System.IO.Compression; -using System.Text; -using System.Text.Json; -using Microsoft.EntityFrameworkCore; -using Microsoft.Extensions.DependencyInjection; -using Tiku.Application.Assets; -using Tiku.Application.Content; -using Tiku.Application.Security; -using Tiku.Application.Storage; -using Tiku.Application.Tenancy; -using Tiku.Domain.Commerce; -using Tiku.Domain.Content; -using Tiku.Domain.Operations; -using Tiku.Domain.Tenancy; -using Tiku.Infrastructure.Persistence; - -namespace Tiku.Infrastructure.Jobs; - -internal sealed partial class BackgroundJobService -{ - private async Task ProcessCoreAsync( - IServiceProvider scopedProvider, - BackgroundJob job, - CancellationToken cancellationToken) - { - cancellationToken.ThrowIfCancellationRequested(); - var scopedDbContext = scopedProvider.GetRequiredService(); - var handlers = new Dictionary>>(StringComparer.Ordinal) - { - ["content_export"] = () => ProcessContentExportAsync(scopedDbContext, job, cancellationToken), - ["content_import"] = () => ProcessContentImportAsync(scopedProvider, job, cancellationToken), - ["asset_security_scan"] = () => - ProcessAssetSecurityScanAsync(scopedProvider, scopedDbContext, job, cancellationToken), - ["tenant_export"] = () => ProcessTenantExportAsync(scopedProvider, scopedDbContext, job, cancellationToken), - ["statistics_aggregation"] = () => - ProcessStatisticsAggregationAsync(scopedProvider, scopedDbContext, job, cancellationToken), - ["commerce_reconciliation"] = - () => ProcessCommerceReconciliationAsync(scopedDbContext, job, cancellationToken), - ["tenant_domain_recheck"] = () => ProcessTenantDomainRecheckAsync(scopedProvider, cancellationToken) - }; - return handlers.TryGetValue(job.JobType, out var handler) - ? await handler() - : 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 static async Task ProcessAssetSecurityScanAsync( - IServiceProvider scopedProvider, - TikuDbContext scopedDbContext, - BackgroundJob job, - CancellationToken cancellationToken) - { - var assetId = GetJsonGuid(job.Payload, "assetId") ?? - throw new InvalidOperationException("asset_security_scan job requires assetId."); - var asset = await scopedDbContext.ContentAssets.SingleOrDefaultAsync( - item => item.TenantId == job.TenantId && item.Id == assetId, - cancellationToken) ?? throw new InvalidOperationException("Asset security scan target was not found."); - if (asset.UploadStatus != AssetUploadStatus.Verified || - string.IsNullOrWhiteSpace(asset.Bucket) || - string.IsNullOrWhiteSpace(asset.ObjectKey)) - throw new InvalidOperationException("Asset must have a verified object location before security scanning."); - - asset.SecurityScanStatus = AssetSecurityScanStatus.Scanning; - await scopedDbContext.SaveChangesAsync(cancellationToken); - var scanner = scopedProvider.GetRequiredService(); - var storage = scopedProvider.GetRequiredService(); - await using var content = await storage.OpenReadAsync( - new ObjectStorageReadRequest( - job.TenantId, - asset.StorageProvider switch - { - AssetStorageProvider.AliyunOss => ObjectStorageProviders.AliyunOss, - AssetStorageProvider.LocalDev => ObjectStorageProviders.LocalDev, - _ => throw new InvalidOperationException( - "Asset storage provider does not support security scanning.") - }, - asset.Bucket, - asset.ObjectKey), - cancellationToken); - var result = - await scanner.ScanAsync(content, asset.VerifiedSizeBytes ?? asset.FileSizeBytes, cancellationToken); - var infected = result.Verdict == AssetSecurityScanVerdict.Infected; - asset.SecurityScanStatus = infected ? AssetSecurityScanStatus.Failed : AssetSecurityScanStatus.Passed; - asset.SecurityScannedAt = DateTimeOffset.UtcNow; - asset.SecurityScanProvider = result.Provider; - asset.SecurityScanSummary = JsonSerializer.SerializeToElement(new - { - verdict = result.Verdict.ToString(), - result.Signature, - result.BytesScanned - }); - scopedDbContext.ContentAssetSecurityScanEvents.Add(new ContentAssetSecurityScanEvent - { - TenantId = job.TenantId, - AssetId = asset.Id, - Provider = result.Provider, - ScanStatus = asset.SecurityScanStatus, - RiskLevel = infected ? AssetSecurityRiskLevel.Critical : AssetSecurityRiskLevel.None, - IssueCodes = infected ? [result.Signature ?? "malware_detected"] : [], - Details = asset.SecurityScanSummary - }); - await scopedDbContext.SaveChangesAsync(cancellationToken); - return JsonSerializer.SerializeToElement(new - { - assetId = asset.Id, - status = asset.SecurityScanStatus.ToString(), - result.Signature, - result.BytesScanned - }); - } - - private async Task RecordAssetScanRetryAsync( - BackgroundJob job, - AssetSecurityScannerException exception, - CancellationToken cancellationToken) - { - var assetId = GetJsonGuid(job.Payload, "assetId"); - if (assetId is null) return; - - var asset = await dbContext.ContentAssets.SingleOrDefaultAsync( - item => item.TenantId == job.TenantId && item.Id == assetId, - cancellationToken); - if (asset is null) return; - - asset.SecurityScanStatus = AssetSecurityScanStatus.Pending; - asset.SecurityScanProvider = "clamav"; - asset.SecurityScanSummary = JsonSerializer.SerializeToElement(new { errorCode = exception.Code }); - dbContext.ContentAssetSecurityScanEvents.Add(new ContentAssetSecurityScanEvent - { - TenantId = job.TenantId, - AssetId = asset.Id, - Provider = "clamav", - ScanStatus = AssetSecurityScanStatus.Pending, - RiskLevel = AssetSecurityRiskLevel.None, - IssueCodes = [exception.Code], - Details = asset.SecurityScanSummary - }); - await dbContext.SaveChangesAsync(cancellationToken); - } - - private static async Task ProcessTenantExportAsync( - IServiceProvider scopedProvider, - TikuDbContext scopedDbContext, - BackgroundJob job, - CancellationToken cancellationToken) - { - var operationId = GetJsonGuid(job.Payload, "operationId") ?? - throw new InvalidOperationException("tenant_export job requires operationId."); - var operation = await scopedDbContext.TenantLifecycleOperations.SingleOrDefaultAsync(item => - item.TenantId == job.TenantId && item.Id == operationId && - item.OperationType == TenantLifecycleOperationType.Export, - cancellationToken) ?? throw new InvalidOperationException("Tenant export operation was not found."); - operation.Status = TenantLifecycleOperationStatus.Processing; - operation.StartedAt ??= DateTimeOffset.UtcNow; - operation.LastError = null; - await scopedDbContext.SaveChangesAsync(cancellationToken); - - var temporaryPath = Path.Combine(Path.GetTempPath(), $"tiku-tenant-export-{operation.Id:N}.tar.gz"); - try - { - var tenant = await scopedDbContext.Tenants.AsNoTracking() - .SingleAsync(item => item.Id == job.TenantId, cancellationToken); - var memberships = await scopedDbContext.TenantMemberships.AsNoTracking() - .Where(item => item.TenantId == job.TenantId) - .Select(item => new { item.UserId, item.Role, item.Status, item.CreatedAt, item.UpdatedAt }) - .ToArrayAsync(cancellationToken); - var domains = await scopedDbContext.TenantDomains.AsNoTracking() - .Where(item => item.TenantId == job.TenantId) - .Select(item => new - { - item.Id, item.Host, item.DomainType, item.Status, item.IsPrimary, item.CreatedAt, item.UpdatedAt - }) - .ToArrayAsync(cancellationToken); - var assets = await scopedDbContext.ContentAssets.AsNoTracking() - .Where(item => item.TenantId == job.TenantId && item.Status == ContentStatus.Active) - .ToArrayAsync(cancellationToken); - var storage = scopedProvider.GetRequiredService(); - - await using (var file = new FileStream(temporaryPath, FileMode.CreateNew, FileAccess.Write, FileShare.None, - 128 * 1024, FileOptions.Asynchronous)) - await using (var gzip = new GZipStream(file, CompressionLevel.Fastest, false)) - await using (var archive = new TarWriter(gzip, TarEntryFormat.Pax, false)) - { - await WriteJsonEntryAsync(archive, "manifest.json", new - { - format = "tiku-tenant-export", - version = 1, - tenantId = job.TenantId, - operationId, - generatedAt = DateTimeOffset.UtcNow, - exclusions = new[] - { - "password_hashes", "auth_tokens", "refresh_tokens", "secret_plaintext", "data_protection_keys", - "global_platform_data" - }, - tables = new[] { "tenant", "tenant_memberships", "tenant_domains", "content_assets" } - }, cancellationToken); - await WriteJsonLinesEntryAsync(archive, "data/tenant.jsonl", new[] - { - new - { - tenant.Id, tenant.Slug, tenant.Name, tenant.LegalName, tenant.Status, tenant.Mode, - tenant.BillingStatus, tenant.OwnerUserId, tenant.CreatedAt, tenant.UpdatedAt - } - }, cancellationToken); - await WriteJsonLinesEntryAsync(archive, "data/tenant_memberships.jsonl", memberships, - cancellationToken); - await WriteJsonLinesEntryAsync(archive, "data/tenant_domains.jsonl", domains, cancellationToken); - await WriteJsonLinesEntryAsync(archive, "data/content_assets.jsonl", assets.Select(item => new - { - item.Id, - item.AssetKey, - item.Title, - item.FileName, - item.StorageProvider, - item.Bucket, - item.ObjectKey, - item.MimeType, - item.FileSizeBytes, - item.ChecksumSha256, - item.UploadStatus, - item.SecurityScanStatus, - item.CreatedAt, - item.UpdatedAt - }), cancellationToken); - - foreach (var asset in assets.Where(item => - item.UploadStatus == AssetUploadStatus.Verified && - item.SecurityScanStatus is AssetSecurityScanStatus.Passed - or AssetSecurityScanStatus.NotRequired && - !string.IsNullOrWhiteSpace(item.Bucket) && - !string.IsNullOrWhiteSpace(item.ObjectKey))) - { - var provider = asset.StorageProvider switch - { - AssetStorageProvider.AliyunOss => ObjectStorageProviders.AliyunOss, - AssetStorageProvider.LocalDev => ObjectStorageProviders.LocalDev, - _ => null - }; - if (provider is null) continue; - await using var content = await storage.OpenReadAsync( - new ObjectStorageReadRequest( - job.TenantId, provider, asset.Bucket!, asset.ObjectKey!), - cancellationToken); - var name = SanitizeTarPath(asset.FileName ?? asset.Id.ToString("N")); - archive.WriteEntry(new PaxTarEntry(TarEntryType.RegularFile, $"assets/{asset.Id:N}/{name}") - { - DataStream = content - }); - } - } - - await using var upload = new FileStream(temporaryPath, FileMode.Open, FileAccess.Read, FileShare.Read, - 128 * 1024, FileOptions.Asynchronous); - var providerName = storage.ConfiguredDefaultProvider(); - var bucket = storage.ConfiguredDefaultBucket(); - var objectKey = - storage.ValidateObjectKey(job.TenantId, $"{job.TenantId:N}/tenant-exports/{operation.Id:N}.tar.gz"); - var written = await storage.WriteObjectAsync( - new ObjectStorageWriteRequest( - job.TenantId, - providerName, - bucket, - objectKey, - "application/gzip", - upload, - upload.Length, - Upsert: false), - cancellationToken); - var exportAsset = new ContentAsset - { - TenantId = job.TenantId, - Title = "Tenant export archive", - FileName = $"tenant-export-{operation.Id:N}.tar.gz", - AssetType = ContentAssetType.Document, - StorageProvider = providerName switch - { - ObjectStorageProviders.AliyunOss => AssetStorageProvider.AliyunOss, - ObjectStorageProviders.LocalDev => AssetStorageProvider.LocalDev, - _ => AssetStorageProvider.ExternalUrl - }, - Bucket = written.Bucket, - ObjectKey = written.ObjectKey, - MimeType = "application/gzip", - FileSizeBytes = written.SizeBytes, - ChecksumSha256 = written.ChecksumSha256, - UploadStatus = AssetUploadStatus.Verified, - VerifiedAt = DateTimeOffset.UtcNow, - VerifiedSizeBytes = written.SizeBytes, - SecurityScanStatus = AssetSecurityScanStatus.NotRequired, - Source = "tenant_export" - }; - scopedDbContext.ContentAssets.Add(exportAsset); - operation.ExportAssetId = exportAsset.Id; - operation.Status = TenantLifecycleOperationStatus.Succeeded; - operation.CompletedAt = DateTimeOffset.UtcNow; - operation.Result = JsonSerializer.SerializeToElement(new - { - exportAssetId = exportAsset.Id, - written.SizeBytes, - assetCount = assets.Length - }); - await scopedDbContext.SaveChangesAsync(cancellationToken); - job.OutputAssetId = exportAsset.Id; - return operation.Result; - } - catch (Exception exception) when (exception is not OperationCanceledException) - { - operation.Status = TenantLifecycleOperationStatus.Failed; - operation.LastError = exception.Message; - operation.CompletedAt = DateTimeOffset.UtcNow; - await scopedDbContext.SaveChangesAsync(cancellationToken); - throw; - } - finally - { - if (File.Exists(temporaryPath)) File.Delete(temporaryPath); - } - } - - private static async Task WriteJsonEntryAsync( - TarWriter archive, - string name, - T value, - CancellationToken cancellationToken) - { - var stream = new MemoryStream(); - await JsonSerializer.SerializeAsync(stream, value, cancellationToken: cancellationToken); - stream.Position = 0; - archive.WriteEntry(new PaxTarEntry(TarEntryType.RegularFile, name) { DataStream = stream }); - await stream.DisposeAsync(); - } - - private static async Task WriteJsonLinesEntryAsync( - TarWriter archive, - string name, - IEnumerable values, - CancellationToken cancellationToken) - { - var stream = new MemoryStream(); - await using (var writer = new StreamWriter(stream, new UTF8Encoding(false), leaveOpen: true)) - { - foreach (var value in values) - { - cancellationToken.ThrowIfCancellationRequested(); - await writer.WriteLineAsync(JsonSerializer.Serialize(value)); - } - } - - stream.Position = 0; - archive.WriteEntry(new PaxTarEntry(TarEntryType.RegularFile, name) { DataStream = stream }); - await stream.DisposeAsync(); - } - - private static string SanitizeTarPath(string value) - { - var name = Path.GetFileName(value).Replace('\\', '_').Replace('/', '_'); - return string.IsNullOrWhiteSpace(name) ? "asset.bin" : name; - } - - 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 == TenantExternalProviderCapability.Payment && - item.Provider == provider && - item.Status == 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 - }); - } -} \ No newline at end of file diff --git a/Tiku.Infrastructure/Jobs/Processor/BackgroundJobService.Processor.cs b/Tiku.Infrastructure/Jobs/Processor/BackgroundJobService.Processor.cs index e59dd02..15cdfbd 100644 --- a/Tiku.Infrastructure/Jobs/Processor/BackgroundJobService.Processor.cs +++ b/Tiku.Infrastructure/Jobs/Processor/BackgroundJobService.Processor.cs @@ -1,7 +1,8 @@ using System.Diagnostics; using System.Text.Json; using Microsoft.EntityFrameworkCore; -using Tiku.Application.Assets; +using Microsoft.Extensions.DependencyInjection; +using Tiku.Application.Jobs; using Tiku.Application.Security; using Tiku.Domain.Common; using Tiku.Domain.Operations; @@ -137,11 +138,18 @@ internal sealed partial class BackgroundJobService try { - var result = await tenantExecutionScope.ExecuteAsync( + var handlerResult = await tenantExecutionScope.ExecuteAsync( new SystemScopeRequest( job.TenantId, SystemScopeCallerType.Worker, workerId, $"Background job {job.JobType}", job.Id.ToString("N")), - (provider, token) => ProcessCoreAsync(provider, job, token), + async (provider, token) => + { + var registry = provider.GetRequiredService(); + var handler = registry.GetRequired(job.JobType); + return await handler.HandleAsync( + new BackgroundJobExecutionContext(job.Id, job.TenantId, job.JobType, job.Payload), + token); + }, cancellationToken); var cancellationRequested = await dbContext.BackgroundJobs.AsNoTracking() .Where(item => item.Id == job.Id) @@ -150,22 +158,42 @@ internal sealed partial class BackgroundJobService job.Status = cancellationRequested ? BackgroundJobStatus.Cancelled : BackgroundJobStatus.Succeeded; job.CompletedAt = DateTimeOffset.UtcNow; job.LastError = null; - job.Result = result; + job.Result = handlerResult.Result; + job.OutputAssetId = handlerResult.OutputAssetId; } catch (Exception exception) when (exception is not OperationCanceledException) { job.RetryCount++; - job.LastError = exception is AssetSecurityScannerException scannerException - ? $"{scannerException.Code}: {scannerException.Message}" + job.LastError = exception is BackgroundJobHandlerException handlerException + ? $"{handlerException.Code}: {handlerException.Message}" : exception.Message; - if (exception is AssetSecurityScannerException assetScanException) - await RecordAssetScanRetryAsync(job, assetScanException, cancellationToken); 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; + try + { + await tenantExecutionScope.ExecuteAsync( + new SystemScopeRequest( + job.TenantId, SystemScopeCallerType.Worker, workerId, + $"Background job failure compensation {job.JobType}", job.Id.ToString("N")), + async (provider, token) => + { + var handler = provider.GetRequiredService() + .GetRequired(job.JobType); + await handler.HandleFailureAsync( + new BackgroundJobExecutionContext(job.Id, job.TenantId, job.JobType, job.Payload), + exception, + token); + }, + cancellationToken); + } + catch (Exception compensationException) when (compensationException is not OperationCanceledException) + { + job.LastError = $"{job.LastError} Compensation failed: {compensationException.Message}"; + } } finally { @@ -199,4 +227,4 @@ internal sealed partial class BackgroundJobService .SetProperty(value => value.LockExpiresAt, (DateTimeOffset?)null), cancellationToken); } -} \ No newline at end of file +} diff --git a/Tiku.Infrastructure/Modules/CommerceModule.cs b/Tiku.Infrastructure/Modules/CommerceModule.cs index 680b30e..f3ca031 100644 --- a/Tiku.Infrastructure/Modules/CommerceModule.cs +++ b/Tiku.Infrastructure/Modules/CommerceModule.cs @@ -1,9 +1,11 @@ using Microsoft.Extensions.DependencyInjection; using Tiku.Application.Commerce; using Tiku.Application.Growth; +using Tiku.Application.Jobs; using Tiku.Application.Notifications; using Tiku.Application.Points; using Tiku.Infrastructure.Commerce; +using Tiku.Infrastructure.Commerce.Reconciliation; using Tiku.Infrastructure.Growth; using Tiku.Infrastructure.Notifications; using Tiku.Infrastructure.Points; @@ -16,6 +18,7 @@ internal static class CommerceModule { services.AddScoped(); services.AddScoped(); + services.AddScoped(); services.AddScoped(); services.AddScoped(); services.AddScoped(); @@ -31,4 +34,4 @@ internal static class CommerceModule services.AddScoped(); return services; } -} \ No newline at end of file +} diff --git a/Tiku.Infrastructure/Modules/ContentModule.cs b/Tiku.Infrastructure/Modules/ContentModule.cs index 046ff97..4e16c48 100644 --- a/Tiku.Infrastructure/Modules/ContentModule.cs +++ b/Tiku.Infrastructure/Modules/ContentModule.cs @@ -2,6 +2,7 @@ using Microsoft.Extensions.DependencyInjection; using Tiku.Application.Assets; using Tiku.Application.Catalog; using Tiku.Application.Content; +using Tiku.Application.Jobs; using Tiku.Application.Profile; using Tiku.Application.QuestionBanks; using Tiku.Application.Scoreline; @@ -10,6 +11,9 @@ using Tiku.Application.StudyContent; using Tiku.Infrastructure.Assets; using Tiku.Infrastructure.Catalog; using Tiku.Infrastructure.Content; +using Tiku.Infrastructure.Content.Exports; +using Tiku.Infrastructure.Content.Imports; +using Tiku.Infrastructure.Assets.Security; using Tiku.Infrastructure.Profile; using Tiku.Infrastructure.QuestionBanks; using Tiku.Infrastructure.Scoreline; @@ -38,6 +42,9 @@ internal static class ContentModule services.AddScoped(); services.AddScoped(); services.AddScoped(); + services.AddScoped(); + services.AddScoped(); + services.AddScoped(); services.AddOptions(); services.AddScoped(); services.AddOptions() @@ -48,4 +55,4 @@ internal static class ContentModule services.AddSingleton(); return services; } -} \ No newline at end of file +} diff --git a/Tiku.Infrastructure/Modules/JobsModule.cs b/Tiku.Infrastructure/Modules/JobsModule.cs index 7987740..eecc9a6 100644 --- a/Tiku.Infrastructure/Modules/JobsModule.cs +++ b/Tiku.Infrastructure/Modules/JobsModule.cs @@ -13,6 +13,7 @@ internal static class JobsModule services.AddScoped(provider => provider.GetRequiredService()); services.AddScoped(provider => provider.GetRequiredService()); services.AddScoped(provider => provider.GetRequiredService()); + services.AddScoped(); return services; } -} \ No newline at end of file +} diff --git a/Tiku.Infrastructure/Modules/PlatformCoreModule.cs b/Tiku.Infrastructure/Modules/PlatformCoreModule.cs index fcd7024..89c1057 100644 --- a/Tiku.Infrastructure/Modules/PlatformCoreModule.cs +++ b/Tiku.Infrastructure/Modules/PlatformCoreModule.cs @@ -1,5 +1,6 @@ using Microsoft.Extensions.DependencyInjection; using Tiku.Application.Backoffice; +using Tiku.Application.Jobs; using Tiku.Application.Security; using Tiku.Application.Tenancy; using Tiku.Infrastructure.Backoffice; @@ -28,6 +29,7 @@ internal static class PlatformCoreModule services.AddHttpClient(); services.AddHttpClient(); services.AddScoped(); + services.AddScoped(); services.AddOptions(); services.AddSingleton(); services.AddScoped(); @@ -45,4 +47,4 @@ internal static class PlatformCoreModule services.AddScoped(); return services; } -} \ No newline at end of file +} diff --git a/Tiku.Infrastructure/Modules/PlatformModule.cs b/Tiku.Infrastructure/Modules/PlatformModule.cs index b5e57b7..b716b58 100644 --- a/Tiku.Infrastructure/Modules/PlatformModule.cs +++ b/Tiku.Infrastructure/Modules/PlatformModule.cs @@ -2,8 +2,10 @@ using Microsoft.Extensions.DependencyInjection; using Tiku.Application.PlatformAdmin; using Tiku.Application.PlatformAdmin.Operations; using Tiku.Application.PlatformBilling; +using Tiku.Application.Jobs; using Tiku.Infrastructure.PlatformAdmin; using Tiku.Infrastructure.PlatformAdmin.Operations; +using Tiku.Infrastructure.PlatformAdmin.TenantProvisioning; using Tiku.Infrastructure.PlatformBilling; namespace Tiku.Infrastructure; @@ -13,6 +15,7 @@ internal static class PlatformModule internal static IServiceCollection AddPlatformModule(this IServiceCollection services) { services.AddScoped(); + services.AddScoped(); services.AddScoped(); services.AddOptions(); services.AddScoped(); @@ -33,4 +36,4 @@ internal static class PlatformModule services.AddOptions(); return services; } -} \ No newline at end of file +} diff --git a/Tiku.Infrastructure/Modules/TenantAdminModule.cs b/Tiku.Infrastructure/Modules/TenantAdminModule.cs index afcdd05..5ae6264 100644 --- a/Tiku.Infrastructure/Modules/TenantAdminModule.cs +++ b/Tiku.Infrastructure/Modules/TenantAdminModule.cs @@ -1,8 +1,10 @@ using Microsoft.Extensions.DependencyInjection; using Tiku.Application.Tenancy; using Tiku.Application.TenantAdmin; +using Tiku.Application.Jobs; using Tiku.Infrastructure.Tenancy; using Tiku.Infrastructure.TenantAdmin; +using Tiku.Infrastructure.TenantAdmin.Dashboard; namespace Tiku.Infrastructure; @@ -13,6 +15,7 @@ internal static class TenantAdminModule services.AddScoped(); services.AddScoped(); services.AddScoped(); + services.AddScoped(); return services; } -} \ No newline at end of file +} diff --git a/Tiku.Infrastructure/PlatformAdmin/TenantProvisioning/TenantExportJobHandler.cs b/Tiku.Infrastructure/PlatformAdmin/TenantProvisioning/TenantExportJobHandler.cs new file mode 100644 index 0000000..b46e15a --- /dev/null +++ b/Tiku.Infrastructure/PlatformAdmin/TenantProvisioning/TenantExportJobHandler.cs @@ -0,0 +1,244 @@ +using System.Formats.Tar; +using System.IO.Compression; +using System.Text; +using System.Text.Json; +using Microsoft.EntityFrameworkCore; +using Tiku.Application.Jobs; +using Tiku.Application.Storage; +using Tiku.Domain.Content; +using Tiku.Domain.Operations; +using Tiku.Domain.Tenancy; +using Tiku.Infrastructure.Jobs; +using Tiku.Infrastructure.Persistence; + +namespace Tiku.Infrastructure.PlatformAdmin.TenantProvisioning; + +internal sealed class TenantExportJobHandler( + TikuDbContext dbContext, + IObjectStorageService storage) : IBackgroundJobHandler +{ + public string JobType => "tenant_export"; + + public async Task HandleAsync( + BackgroundJobExecutionContext context, + CancellationToken cancellationToken = default) + { + var operationId = BackgroundJobPayload.GetGuid(context.Payload, "operationId") ?? + throw new InvalidOperationException("tenant_export job requires operationId."); + var operation = await dbContext.TenantLifecycleOperations.SingleOrDefaultAsync(item => + item.TenantId == context.TenantId && item.Id == operationId && + item.OperationType == TenantLifecycleOperationType.Export, + cancellationToken) ?? throw new InvalidOperationException("Tenant export operation was not found."); + operation.Status = TenantLifecycleOperationStatus.Processing; + operation.StartedAt ??= DateTimeOffset.UtcNow; + operation.LastError = null; + await dbContext.SaveChangesAsync(cancellationToken); + + var temporaryPath = Path.Combine(Path.GetTempPath(), $"tiku-tenant-export-{operation.Id:N}.tar.gz"); + try + { + var tenant = await dbContext.Tenants.AsNoTracking() + .SingleAsync(item => item.Id == context.TenantId, cancellationToken); + var memberships = await dbContext.TenantMemberships.AsNoTracking() + .Where(item => item.TenantId == context.TenantId) + .Select(item => new { item.UserId, item.Role, item.Status, item.CreatedAt, item.UpdatedAt }) + .ToArrayAsync(cancellationToken); + var domains = await dbContext.TenantDomains.AsNoTracking() + .Where(item => item.TenantId == context.TenantId) + .Select(item => new + { + item.Id, + item.Host, + item.DomainType, + item.Status, + item.IsPrimary, + item.CreatedAt, + item.UpdatedAt + }) + .ToArrayAsync(cancellationToken); + var assets = await dbContext.ContentAssets.AsNoTracking() + .Where(item => item.TenantId == context.TenantId && item.Status == ContentStatus.Active) + .ToArrayAsync(cancellationToken); + + await using (var file = new FileStream(temporaryPath, FileMode.CreateNew, FileAccess.Write, FileShare.None, + 128 * 1024, FileOptions.Asynchronous)) + await using (var gzip = new GZipStream(file, CompressionLevel.Fastest, false)) + await using (var archive = new TarWriter(gzip, TarEntryFormat.Pax, false)) + { + await WriteJsonEntryAsync(archive, "manifest.json", new + { + format = "tiku-tenant-export", + version = 1, + tenantId = context.TenantId, + operationId, + generatedAt = DateTimeOffset.UtcNow, + exclusions = new[] + { + "password_hashes", "auth_tokens", "refresh_tokens", "secret_plaintext", + "data_protection_keys", "global_platform_data" + }, + tables = new[] { "tenant", "tenant_memberships", "tenant_domains", "content_assets" } + }, cancellationToken); + await WriteJsonLinesEntryAsync(archive, "data/tenant.jsonl", new[] + { + new + { + tenant.Id, tenant.Slug, tenant.Name, tenant.LegalName, tenant.Status, tenant.Mode, + tenant.BillingStatus, tenant.OwnerUserId, tenant.CreatedAt, tenant.UpdatedAt + } + }, cancellationToken); + await WriteJsonLinesEntryAsync(archive, "data/tenant_memberships.jsonl", memberships, + cancellationToken); + await WriteJsonLinesEntryAsync(archive, "data/tenant_domains.jsonl", domains, cancellationToken); + await WriteJsonLinesEntryAsync(archive, "data/content_assets.jsonl", assets.Select(item => new + { + item.Id, + item.AssetKey, + item.Title, + item.FileName, + item.StorageProvider, + item.Bucket, + item.ObjectKey, + item.MimeType, + item.FileSizeBytes, + item.ChecksumSha256, + item.UploadStatus, + item.SecurityScanStatus, + item.CreatedAt, + item.UpdatedAt + }), cancellationToken); + + foreach (var asset in assets.Where(item => + item.UploadStatus == AssetUploadStatus.Verified && + item.SecurityScanStatus is AssetSecurityScanStatus.Passed + or AssetSecurityScanStatus.NotRequired && + !string.IsNullOrWhiteSpace(item.Bucket) && + !string.IsNullOrWhiteSpace(item.ObjectKey))) + { + var provider = asset.StorageProvider switch + { + AssetStorageProvider.AliyunOss => ObjectStorageProviders.AliyunOss, + AssetStorageProvider.LocalDev => ObjectStorageProviders.LocalDev, + _ => null + }; + if (provider is null) continue; + await using var content = await storage.OpenReadAsync( + new ObjectStorageReadRequest( + context.TenantId, provider, asset.Bucket!, asset.ObjectKey!), + cancellationToken); + var name = SanitizeTarPath(asset.FileName ?? asset.Id.ToString("N")); + archive.WriteEntry(new PaxTarEntry(TarEntryType.RegularFile, $"assets/{asset.Id:N}/{name}") + { + DataStream = content + }); + } + } + + await using var upload = new FileStream(temporaryPath, FileMode.Open, FileAccess.Read, FileShare.Read, + 128 * 1024, FileOptions.Asynchronous); + var providerName = storage.ConfiguredDefaultProvider(); + var bucket = storage.ConfiguredDefaultBucket(); + var objectKey = + storage.ValidateObjectKey(context.TenantId, + $"{context.TenantId:N}/tenant-exports/{operation.Id:N}.tar.gz"); + var written = await storage.WriteObjectAsync( + new ObjectStorageWriteRequest( + context.TenantId, + providerName, + bucket, + objectKey, + "application/gzip", + upload, + upload.Length, + Upsert: false), + cancellationToken); + var exportAsset = new ContentAsset + { + TenantId = context.TenantId, + Title = "Tenant export archive", + FileName = $"tenant-export-{operation.Id:N}.tar.gz", + AssetType = ContentAssetType.Document, + StorageProvider = providerName switch + { + ObjectStorageProviders.AliyunOss => AssetStorageProvider.AliyunOss, + ObjectStorageProviders.LocalDev => AssetStorageProvider.LocalDev, + _ => AssetStorageProvider.ExternalUrl + }, + Bucket = written.Bucket, + ObjectKey = written.ObjectKey, + MimeType = "application/gzip", + FileSizeBytes = written.SizeBytes, + ChecksumSha256 = written.ChecksumSha256, + UploadStatus = AssetUploadStatus.Verified, + VerifiedAt = DateTimeOffset.UtcNow, + VerifiedSizeBytes = written.SizeBytes, + SecurityScanStatus = AssetSecurityScanStatus.NotRequired, + Source = "tenant_export" + }; + dbContext.ContentAssets.Add(exportAsset); + operation.ExportAssetId = exportAsset.Id; + operation.Status = TenantLifecycleOperationStatus.Succeeded; + operation.CompletedAt = DateTimeOffset.UtcNow; + operation.Result = JsonSerializer.SerializeToElement(new + { + exportAssetId = exportAsset.Id, + written.SizeBytes, + assetCount = assets.Length + }); + await dbContext.SaveChangesAsync(cancellationToken); + return new BackgroundJobHandlerResult(operation.Result, exportAsset.Id); + } + catch (Exception exception) when (exception is not OperationCanceledException) + { + operation.Status = TenantLifecycleOperationStatus.Failed; + operation.LastError = exception.Message; + operation.CompletedAt = DateTimeOffset.UtcNow; + await dbContext.SaveChangesAsync(cancellationToken); + throw; + } + finally + { + if (File.Exists(temporaryPath)) File.Delete(temporaryPath); + } + } + + private static async Task WriteJsonEntryAsync( + TarWriter archive, + string name, + T value, + CancellationToken cancellationToken) + { + var stream = new MemoryStream(); + await JsonSerializer.SerializeAsync(stream, value, cancellationToken: cancellationToken); + stream.Position = 0; + archive.WriteEntry(new PaxTarEntry(TarEntryType.RegularFile, name) { DataStream = stream }); + await stream.DisposeAsync(); + } + + private static async Task WriteJsonLinesEntryAsync( + TarWriter archive, + string name, + IEnumerable values, + CancellationToken cancellationToken) + { + var stream = new MemoryStream(); + await using (var writer = new StreamWriter(stream, new UTF8Encoding(false), leaveOpen: true)) + { + foreach (var value in values) + { + cancellationToken.ThrowIfCancellationRequested(); + await writer.WriteLineAsync(JsonSerializer.Serialize(value)); + } + } + + stream.Position = 0; + archive.WriteEntry(new PaxTarEntry(TarEntryType.RegularFile, name) { DataStream = stream }); + await stream.DisposeAsync(); + } + + private static string SanitizeTarPath(string value) + { + var name = Path.GetFileName(value).Replace('\\', '_').Replace('/', '_'); + return string.IsNullOrWhiteSpace(name) ? "asset.bin" : name; + } +} diff --git a/Tiku.Infrastructure/Tenancy/TenantDomainRecheckJobHandler.cs b/Tiku.Infrastructure/Tenancy/TenantDomainRecheckJobHandler.cs new file mode 100644 index 0000000..e987259 --- /dev/null +++ b/Tiku.Infrastructure/Tenancy/TenantDomainRecheckJobHandler.cs @@ -0,0 +1,19 @@ +using System.Text.Json; +using Tiku.Application.Jobs; +using Tiku.Application.Tenancy; + +namespace Tiku.Infrastructure.Tenancy; + +internal sealed class TenantDomainRecheckJobHandler(ITenantDomainLifecycleService lifecycleService) + : IBackgroundJobHandler +{ + public string JobType => "tenant_domain_recheck"; + + public async Task HandleAsync( + BackgroundJobExecutionContext context, + CancellationToken cancellationToken = default) + { + var processed = await lifecycleService.ProcessPendingAsync(cancellationToken); + return new BackgroundJobHandlerResult(JsonSerializer.SerializeToElement(new { processed })); + } +} diff --git a/Tiku.Infrastructure/TenantAdmin/Dashboard/StatisticsAggregationJobHandler.cs b/Tiku.Infrastructure/TenantAdmin/Dashboard/StatisticsAggregationJobHandler.cs new file mode 100644 index 0000000..0d44f9b --- /dev/null +++ b/Tiku.Infrastructure/TenantAdmin/Dashboard/StatisticsAggregationJobHandler.cs @@ -0,0 +1,52 @@ +using System.Text.Json; +using Microsoft.EntityFrameworkCore; +using Tiku.Application.Jobs; +using Tiku.Application.Security; +using Tiku.Domain.Commerce; +using Tiku.Infrastructure.Persistence; + +namespace Tiku.Infrastructure.TenantAdmin.Dashboard; + +internal sealed class StatisticsAggregationJobHandler( + TikuDbContext dbContext, + IFeatureUsageReconciliationService featureUsageReconciliationService) : IBackgroundJobHandler +{ + public string JobType => "statistics_aggregation"; + + public async Task HandleAsync( + BackgroundJobExecutionContext context, + CancellationToken cancellationToken = default) + { + var since = DateTimeOffset.UtcNow.AddDays(-7); + var activeLearnerCount = await dbContext.PracticeSessions + .Where(item => item.TenantId == context.TenantId && item.StartedAt >= since) + .Select(item => item.UserId) + .Distinct() + .CountAsync(cancellationToken); + var paidOrderCount = await dbContext.Orders + .CountAsync(item => item.TenantId == context.TenantId && item.Status == OrderStatus.Paid, + cancellationToken); + var revenueCents = await dbContext.Orders + .Where(item => item.TenantId == context.TenantId && + (item.Status == OrderStatus.Paid || + item.Status == OrderStatus.PartiallyRefunded || + item.Status == OrderStatus.Refunded)) + .SumAsync(item => item.AmountCents - item.RefundedAmountCents, cancellationToken); + var quotaUsage = await featureUsageReconciliationService.ReconcileTenantAsync( + new ReconcileFeatureUsageRequest( + context.TenantId, + SystemScopeCallerType.Worker, + nameof(StatisticsAggregationJobHandler), + "Reconcile tenant current feature usage during statistics aggregation", + $"feature-usage-{context.JobId:N}"), + cancellationToken); + return new BackgroundJobHandlerResult(JsonSerializer.SerializeToElement(new + { + since, + activeLearnerCount, + paidOrderCount, + revenueCents, + quotaUsage + })); + } +} diff --git a/Tiku.IntegrationTests/ArchitectureBoundaryTests.cs b/Tiku.IntegrationTests/ArchitectureBoundaryTests.cs index 7e99d31..9ccae4d 100644 --- a/Tiku.IntegrationTests/ArchitectureBoundaryTests.cs +++ b/Tiku.IntegrationTests/ArchitectureBoundaryTests.cs @@ -30,34 +30,94 @@ public sealed class ArchitectureBoundaryTests } [Fact] - public void New_service_files_do_not_exceed_the_hard_size_limit() + public void Service_classes_do_not_exceed_the_hard_size_limit_or_grow_legacy_debt() { var root = FindRepositoryRoot(); - var legacyFacades = new HashSet(StringComparer.Ordinal) + var legacyLineBudgets = new Dictionary(StringComparer.Ordinal) { - "TenantAdminDirectService.cs", - "DirectContentService.cs", - "LearningActivityService.cs", - "CommerceAdminService.cs", - "PlatformAdminService.cs", - "BackgroundJobService.cs" + ["TenantAdminDirectService"] = 3190, + ["DirectContentService"] = 2015, + ["CommerceAdminService"] = 1842, + ["LearningActivityService"] = 1781, + ["PlatformAdminService"] = 1719, + ["ContentManagementService"] = 1096, + ["PlatformQuestionBankService"] = 1033, + ["CommerceService"] = 983, + ["AssetManagementService"] = 948, + ["AuthService"] = 876, + ["ReferralService"] = 849, + ["PlatformTenantCapabilitiesService"] = 805 }; - var violations = Directory.EnumerateFiles( + + var serviceGroups = Directory.EnumerateFiles( Path.Combine(root, "Tiku.Infrastructure"), - "*Service.cs", + "*Service*.cs", SearchOption.AllDirectories) .Where(path => !path.Contains( $"{Path.DirectorySeparatorChar}Migrations{Path.DirectorySeparatorChar}", StringComparison.Ordinal)) - .Where(path => !legacyFacades.Contains(Path.GetFileName(path))) - .Select(path => new { path, Lines = File.ReadLines(path).Count() }) - .Where(item => item.Lines > 800) - .Select(item => $"{Path.GetRelativePath(root, item.path)} ({item.Lines} lines)") + .GroupBy(path => Path.GetFileNameWithoutExtension(path).Split('.')[0], StringComparer.Ordinal) + .Select(group => new + { + Service = group.Key, + Lines = group.Sum(path => File.ReadLines(path).Count()), + Files = group.Count() + }) + .ToArray(); + + var violations = serviceGroups + .Where(item => legacyLineBudgets.TryGetValue(item.Service, out var budget) + ? item.Lines > budget + : item.Lines > 800) + .Select(item => legacyLineBudgets.TryGetValue(item.Service, out var budget) + ? $"{item.Service} ({item.Lines} lines across {item.Files} files; legacy budget {budget})" + : $"{item.Service} ({item.Lines} lines across {item.Files} files; hard limit 800)") .ToArray(); Assert.True( violations.Length == 0, - $"Service files over the 800-line hard limit were found:{Environment.NewLine}{string.Join(Environment.NewLine, violations)}"); + $"Service classes exceeded the hard limit or grew legacy debt:{Environment.NewLine}{string.Join(Environment.NewLine, violations)}"); + } + + [Fact] + public void Background_jobs_are_dispatched_by_module_registered_handlers() + { + var root = FindRepositoryRoot(); + var infrastructureRoot = Path.Combine(root, "Tiku.Infrastructure"); + var sourceFiles = Directory.EnumerateFiles(infrastructureRoot, "*.cs", SearchOption.AllDirectories) + .Where(path => !path.Contains( + $"{Path.DirectorySeparatorChar}Persistence{Path.DirectorySeparatorChar}Migrations{Path.DirectorySeparatorChar}", + StringComparison.Ordinal)) + .ToArray(); + var handlerFiles = sourceFiles + .Where(path => File.ReadAllText(path).Contains(": IBackgroundJobHandler", StringComparison.Ordinal)) + .ToArray(); + var expectedJobTypes = new[] + { + "content_export", + "content_import", + "asset_security_scan", + "tenant_export", + "statistics_aggregation", + "commerce_reconciliation", + "tenant_domain_recheck" + }; + + Assert.Equal(expectedJobTypes.Length, handlerFiles.Length); + foreach (var jobType in expectedJobTypes) + Assert.Single(handlerFiles, path => + File.ReadAllText(path).Contains($"JobType => \"{jobType}\"", StringComparison.Ordinal)); + + var moduleRegistrations = Directory + .EnumerateFiles(Path.Combine(infrastructureRoot, "Modules"), "*.cs", SearchOption.AllDirectories) + .Sum(path => File.ReadLines(path).Count(line => + line.Contains("AddScoped>>", jobsSource, StringComparison.Ordinal); } [Fact] @@ -111,7 +171,6 @@ public sealed class ArchitectureBoundaryTests { Path.Combine(root, "Tiku.Infrastructure", "Growth", "CommissionService.cs"), Path.Combine(root, "Tiku.Infrastructure", "Growth", "ReferralService.cs"), - Path.Combine(root, "Tiku.Application", "TenantAdmin", "TenantAdminDirectModels.cs"), Path.Combine(root, "Tiku.Api", "Controllers", "TenantAdminDirectController.cs") }; @@ -373,4 +432,4 @@ public sealed class ArchitectureBoundaryTests violations.Length == 0, $"{failureMessage}:{Environment.NewLine}{string.Join(Environment.NewLine, violations)}"); } -} \ No newline at end of file +}