refactor(jobs): modularize background job handlers
Some checks failed
ci / release-gate (push) Has been cancelled
Some checks failed
ci / release-gate (push) Has been cancelled
This commit is contained in:
@@ -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<JsonElement> HandleAsync(
|
||||
Task<BackgroundJobHandlerResult> HandleAsync(
|
||||
BackgroundJobExecutionContext context,
|
||||
CancellationToken cancellationToken = default);
|
||||
|
||||
Task HandleFailureAsync(
|
||||
BackgroundJobExecutionContext context,
|
||||
Exception exception,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
@@ -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<BackgroundJobHandlerResult> 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);
|
||||
}
|
||||
}
|
||||
@@ -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<BackgroundJobHandlerResult> 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;
|
||||
}
|
||||
}
|
||||
@@ -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<BackgroundJobHandlerResult> 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);
|
||||
}
|
||||
}
|
||||
@@ -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<BackgroundJobHandlerResult> 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
|
||||
}));
|
||||
}
|
||||
}
|
||||
@@ -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<JsonElement> 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<TEnum>(JsonElement element, string propertyName, TEnum fallback)
|
||||
where TEnum : struct
|
||||
{
|
||||
var value = GetJsonString(element, propertyName);
|
||||
return Enum.TryParse<TEnum>(value, true, out var parsed) ? parsed : fallback;
|
||||
}
|
||||
|
||||
private static BackgroundJobItem ToItem(BackgroundJob job)
|
||||
{
|
||||
return new BackgroundJobItem(
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
using Tiku.Application.Jobs;
|
||||
|
||||
namespace Tiku.Infrastructure.Jobs;
|
||||
|
||||
internal sealed class BackgroundJobHandlerRegistry
|
||||
{
|
||||
private readonly IReadOnlyDictionary<string, IBackgroundJobHandler> handlers;
|
||||
|
||||
public BackgroundJobHandlerRegistry(IEnumerable<IBackgroundJobHandler> handlers)
|
||||
{
|
||||
var registered = new Dictionary<string, IBackgroundJobHandler>(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}'.");
|
||||
}
|
||||
}
|
||||
49
Tiku.Infrastructure/Jobs/Handlers/BackgroundJobPayload.cs
Normal file
49
Tiku.Infrastructure/Jobs/Handlers/BackgroundJobPayload.cs
Normal file
@@ -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<JsonElement> 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<TEnum>(JsonElement element, string propertyName, TEnum fallback)
|
||||
where TEnum : struct
|
||||
{
|
||||
var value = GetString(element, propertyName);
|
||||
return Enum.TryParse<TEnum>(value, true, out var parsed) ? parsed : fallback;
|
||||
}
|
||||
}
|
||||
@@ -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<JsonElement> ProcessCoreAsync(
|
||||
IServiceProvider scopedProvider,
|
||||
BackgroundJob job,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
var scopedDbContext = scopedProvider.GetRequiredService<TikuDbContext>();
|
||||
var handlers = new Dictionary<string, Func<Task<JsonElement>>>(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<JsonElement> ProcessContentImportAsync(
|
||||
IServiceProvider scopedProvider,
|
||||
BackgroundJob job,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var directContentService = scopedProvider.GetRequiredService<IDirectContentService>();
|
||||
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<JsonElement> 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<IAssetSecurityScanner>();
|
||||
var storage = scopedProvider.GetRequiredService<IObjectStorageService>();
|
||||
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<JsonElement> 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<IObjectStorageService>();
|
||||
|
||||
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<T>(
|
||||
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<T>(
|
||||
TarWriter archive,
|
||||
string name,
|
||||
IEnumerable<T> 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<JsonElement> 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<JsonElement> 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<JsonElement> ProcessTenantDomainRecheckAsync(
|
||||
IServiceProvider scopedProvider,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var lifecycleService = scopedProvider.GetRequiredService<ITenantDomainLifecycleService>();
|
||||
var processed = await lifecycleService.ProcessPendingAsync(cancellationToken);
|
||||
return JsonSerializer.SerializeToElement(new
|
||||
{
|
||||
processed
|
||||
});
|
||||
}
|
||||
|
||||
private async Task<JsonElement> 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<IFeatureUsageReconciliationService>()
|
||||
.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
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -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<BackgroundJobHandlerRegistry>();
|
||||
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<BackgroundJobHandlerRegistry>()
|
||||
.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
|
||||
{
|
||||
|
||||
@@ -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<ICommerceService, CommerceService>();
|
||||
services.AddScoped<ICommerceAdminService, CommerceAdminService>();
|
||||
services.AddScoped<IBackgroundJobHandler, CommerceReconciliationJobHandler>();
|
||||
services.AddScoped<IPointService, PointService>();
|
||||
services.AddScoped<IReferralQrcodeGenerator, ReferralQrcodeGenerator>();
|
||||
services.AddScoped<IReferralService, ReferralService>();
|
||||
|
||||
@@ -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<IAssetAccessService, AssetAccessService>();
|
||||
services.AddScoped<IAssetManagementService, AssetManagementService>();
|
||||
services.AddScoped<IAssetSecurityScanner, ClamAvAssetSecurityScanner>();
|
||||
services.AddScoped<IBackgroundJobHandler, ContentImportJobHandler>();
|
||||
services.AddScoped<IBackgroundJobHandler, ContentExportJobHandler>();
|
||||
services.AddScoped<IBackgroundJobHandler, AssetSecurityScanJobHandler>();
|
||||
services.AddOptions<ClamAvOptions>();
|
||||
services.AddScoped<IVideoPlaybackService, VideoPlaybackService>();
|
||||
services.AddOptions<AliyunOssOptions>()
|
||||
|
||||
@@ -13,6 +13,7 @@ internal static class JobsModule
|
||||
services.AddScoped<IBackgroundJobQueue>(provider => provider.GetRequiredService<BackgroundJobService>());
|
||||
services.AddScoped<IBackgroundJobProcessor>(provider => provider.GetRequiredService<BackgroundJobService>());
|
||||
services.AddScoped<IBackgroundJobOperations>(provider => provider.GetRequiredService<BackgroundJobService>());
|
||||
services.AddScoped<BackgroundJobHandlerRegistry>();
|
||||
return services;
|
||||
}
|
||||
}
|
||||
@@ -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<IDomainOwnershipVerifier, DnsDomainOwnershipVerifier>();
|
||||
services.AddHttpClient<IDomainGatewayProvisioner, HttpDomainGatewayProvisioner>();
|
||||
services.AddScoped<ITenantDomainLifecycleService, TenantDomainLifecycleService>();
|
||||
services.AddScoped<IBackgroundJobHandler, TenantDomainRecheckJobHandler>();
|
||||
services.AddOptions<DomainLifecycleOptions>();
|
||||
services.AddSingleton<ITenantExecutionScope, TenantExecutionScope>();
|
||||
services.AddScoped<ICurrentAccessContext, CurrentAccessContext>();
|
||||
|
||||
@@ -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<IPlatformAdminService, PlatformAdminService>();
|
||||
services.AddScoped<IBackgroundJobHandler, TenantExportJobHandler>();
|
||||
services.AddScoped<IPlatformOperationsQueryService, PlatformOperationsQueryService>();
|
||||
services.AddOptions<TenantProvisioningOptions>();
|
||||
services.AddScoped<IPlatformQuestionBankService, PlatformQuestionBankService>();
|
||||
|
||||
@@ -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<ITenantAdminDirectService, TenantAdminDirectService>();
|
||||
services.AddScoped<ITenantOnboardingService, TenantOnboardingService>();
|
||||
services.AddScoped<ITenantLifecycleService, TenantLifecycleService>();
|
||||
services.AddScoped<IBackgroundJobHandler, StatisticsAggregationJobHandler>();
|
||||
return services;
|
||||
}
|
||||
}
|
||||
@@ -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<BackgroundJobHandlerResult> 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<T>(
|
||||
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<T>(
|
||||
TarWriter archive,
|
||||
string name,
|
||||
IEnumerable<T> 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;
|
||||
}
|
||||
}
|
||||
19
Tiku.Infrastructure/Tenancy/TenantDomainRecheckJobHandler.cs
Normal file
19
Tiku.Infrastructure/Tenancy/TenantDomainRecheckJobHandler.cs
Normal file
@@ -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<BackgroundJobHandlerResult> HandleAsync(
|
||||
BackgroundJobExecutionContext context,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var processed = await lifecycleService.ProcessPendingAsync(cancellationToken);
|
||||
return new BackgroundJobHandlerResult(JsonSerializer.SerializeToElement(new { processed }));
|
||||
}
|
||||
}
|
||||
@@ -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<BackgroundJobHandlerResult> 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
|
||||
}));
|
||||
}
|
||||
}
|
||||
@@ -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<string>(StringComparer.Ordinal)
|
||||
var legacyLineBudgets = new Dictionary<string, int>(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<IBackgroundJobHandler", StringComparison.Ordinal)));
|
||||
Assert.Equal(expectedJobTypes.Length, moduleRegistrations);
|
||||
|
||||
var jobsSource = string.Join(Environment.NewLine, Directory
|
||||
.EnumerateFiles(Path.Combine(infrastructureRoot, "Jobs"), "*.cs", SearchOption.AllDirectories)
|
||||
.Select(File.ReadAllText));
|
||||
Assert.DoesNotContain("Dictionary<string, Func<Task<JsonElement>>>", 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")
|
||||
};
|
||||
|
||||
|
||||
Reference in New Issue
Block a user