1088 lines
47 KiB
C#
1088 lines
47 KiB
C#
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using System.Text.Json;
|
|
using System.Formats.Tar;
|
|
using System.IO.Compression;
|
|
using Tiku.Application.Assets;
|
|
using Tiku.Application.Content;
|
|
using Tiku.Application.Jobs;
|
|
using Tiku.Application.Security;
|
|
using Tiku.Application.Tenancy;
|
|
using Tiku.Domain.Commerce;
|
|
using Tiku.Domain.Common;
|
|
using Tiku.Domain.Content;
|
|
using Tiku.Domain.Operations;
|
|
using Tiku.Infrastructure.Persistence;
|
|
using Tiku.Infrastructure.Observability;
|
|
using System.Diagnostics;
|
|
|
|
namespace Tiku.Infrastructure.Jobs;
|
|
|
|
internal sealed class BackgroundJobService(
|
|
TikuDbContext dbContext,
|
|
ITenantExecutionScope tenantExecutionScope,
|
|
IFeatureAccessService featureAccessService) : IBackgroundJobService
|
|
{
|
|
private static readonly TimeSpan LeaseDuration = TimeSpan.FromMinutes(5);
|
|
|
|
public async Task<BackgroundJobItem> EnqueueAsync(
|
|
CreateBackgroundJobCommand command,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
var normalizedJobType = NormalizeJobType(command.JobType);
|
|
var idempotencyKey = NormalizeIdempotencyKey(command.IdempotencyKey);
|
|
if (idempotencyKey is not null)
|
|
{
|
|
var existing = await dbContext.BackgroundJobs.AsNoTracking().SingleOrDefaultAsync(
|
|
item => item.TenantId == command.TenantId && item.JobType == normalizedJobType &&
|
|
item.IdempotencyKey == idempotencyKey,
|
|
cancellationToken);
|
|
if (existing is not null)
|
|
{
|
|
return ToItem(existing);
|
|
}
|
|
}
|
|
|
|
if (!command.IsSystemJob && !(await featureAccessService.EvaluateAsync(
|
|
command.TenantId,
|
|
ResolveRequiredFeature(normalizedJobType, command.Payload),
|
|
FeatureAccessOperation.Write,
|
|
cancellationToken)).Allowed)
|
|
{
|
|
throw new InvalidOperationException("Tenant feature entitlement does not allow this background job.");
|
|
}
|
|
var quotaMetric = command.IsSystemJob ? null : ResolveQuotaMetric(normalizedJobType);
|
|
var quotaConsumed = false;
|
|
if (quotaMetric is not null)
|
|
{
|
|
var quota = (await featureAccessService.GetQuotaSummaryAsync(command.TenantId, cancellationToken))
|
|
.SingleOrDefault(value => value.MetricCode == quotaMetric);
|
|
if (quota is not null)
|
|
{
|
|
quotaConsumed = await featureAccessService.TryConsumeQuotaAsync(
|
|
command.TenantId, quotaMetric, 1, cancellationToken);
|
|
if (!quotaConsumed)
|
|
{
|
|
throw new FeatureAccessException("The background job quota has been exhausted.", "feature_quota_exhausted");
|
|
}
|
|
}
|
|
}
|
|
var job = new BackgroundJob
|
|
{
|
|
TenantId = command.TenantId,
|
|
JobType = normalizedJobType,
|
|
IdempotencyKey = idempotencyKey,
|
|
Payload = command.Payload,
|
|
RunAfter = command.RunAfter,
|
|
MaxRetries = Math.Clamp(command.MaxRetries, 0, 20)
|
|
};
|
|
dbContext.BackgroundJobs.Add(job);
|
|
try
|
|
{
|
|
await dbContext.SaveChangesAsync(cancellationToken);
|
|
}
|
|
catch (DbUpdateException) when (idempotencyKey is not null)
|
|
{
|
|
dbContext.ChangeTracker.Clear();
|
|
var existing = await dbContext.BackgroundJobs.AsNoTracking().SingleOrDefaultAsync(
|
|
item => item.TenantId == command.TenantId && item.JobType == normalizedJobType &&
|
|
item.IdempotencyKey == idempotencyKey,
|
|
cancellationToken);
|
|
if (existing is not null)
|
|
{
|
|
if (quotaConsumed && quotaMetric is not null)
|
|
{
|
|
await featureAccessService.ReleaseQuotaAsync(command.TenantId, quotaMetric, 1, CancellationToken.None);
|
|
}
|
|
return ToItem(existing);
|
|
}
|
|
throw;
|
|
}
|
|
catch
|
|
{
|
|
if (quotaConsumed && quotaMetric is not null)
|
|
{
|
|
await featureAccessService.ReleaseQuotaAsync(command.TenantId, quotaMetric, 1, CancellationToken.None);
|
|
}
|
|
throw;
|
|
}
|
|
return ToItem(job);
|
|
}
|
|
|
|
private static string? ResolveQuotaMetric(string jobType) => jobType switch
|
|
{
|
|
"content_import" => SaasQuotaMetricCatalog.ImportCount,
|
|
"content_export" => SaasQuotaMetricCatalog.ExportCount,
|
|
_ => null
|
|
};
|
|
|
|
public async Task<int> ProcessPendingAsync(
|
|
string workerId,
|
|
int batchSize,
|
|
bool includeImmediateJobs = true,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
var now = DateTimeOffset.UtcNow;
|
|
var leaseExpiresAt = now.Add(LeaseDuration);
|
|
var claimedIds = await dbContext.Database.SqlQuery<Guid>($"""
|
|
UPDATE background_jobs AS job
|
|
SET status = 'processing',
|
|
locked_by = {workerId},
|
|
lock_expires_at = {leaseExpiresAt},
|
|
started_at = COALESCE(started_at, {now}),
|
|
updated_at = {now}
|
|
WHERE job.id IN (
|
|
SELECT candidate.id
|
|
FROM background_jobs AS candidate
|
|
WHERE (
|
|
(candidate.status = 'pending' AND ({includeImmediateJobs} OR candidate.run_after IS NOT NULL) AND
|
|
(candidate.run_after IS NULL OR candidate.run_after <= {now})) OR
|
|
(candidate.status = 'processing' AND candidate.lock_expires_at <= {now})
|
|
)
|
|
ORDER BY candidate.created_at, candidate.id
|
|
FOR UPDATE SKIP LOCKED
|
|
LIMIT {Math.Clamp(batchSize, 1, 100)}
|
|
)
|
|
RETURNING job.id AS "Value"
|
|
""")
|
|
.ToArrayAsync(cancellationToken);
|
|
|
|
var processed = 0;
|
|
dbContext.ChangeTracker.Clear();
|
|
foreach (var jobId in claimedIds)
|
|
{
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
var job = await dbContext.BackgroundJobs.SingleAsync(value => value.Id == jobId, cancellationToken);
|
|
if (await ProcessJobAsync(job, workerId, alreadyClaimed: true, cancellationToken)) processed++;
|
|
dbContext.ChangeTracker.Clear();
|
|
}
|
|
|
|
return processed;
|
|
}
|
|
|
|
public async Task<bool> ProcessRequestedAsync(
|
|
Guid jobId,
|
|
Guid tenantId,
|
|
string jobType,
|
|
string workerId,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
var normalizedJobType = NormalizeJobType(jobType);
|
|
var claimed = await dbContext.BackgroundJobs
|
|
.Where(item => item.Id == jobId && item.TenantId == tenantId &&
|
|
item.JobType == normalizedJobType && item.Status == BackgroundJobStatus.Pending &&
|
|
item.RunAfter == null)
|
|
.ExecuteUpdateAsync(setters => setters
|
|
.SetProperty(item => item.Status, BackgroundJobStatus.Processing)
|
|
.SetProperty(item => item.LockedBy, workerId)
|
|
.SetProperty(item => item.LockExpiresAt, DateTimeOffset.UtcNow.Add(LeaseDuration))
|
|
.SetProperty(item => item.StartedAt, DateTimeOffset.UtcNow), cancellationToken);
|
|
if (claimed == 0)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
dbContext.ChangeTracker.Clear();
|
|
var job = await dbContext.BackgroundJobs.SingleOrDefaultAsync(
|
|
item => item.Id == jobId && item.TenantId == tenantId,
|
|
cancellationToken);
|
|
if (job is null)
|
|
{
|
|
throw new InvalidOperationException("The requested background job does not exist in the target tenant.");
|
|
}
|
|
if (!string.Equals(job.JobType, normalizedJobType, StringComparison.Ordinal))
|
|
{
|
|
throw new InvalidOperationException("The requested background job type does not match the persisted job.");
|
|
}
|
|
return await ProcessJobAsync(job, workerId, alreadyClaimed: true, cancellationToken);
|
|
}
|
|
|
|
private async Task<bool> ProcessJobAsync(
|
|
BackgroundJob job,
|
|
string workerId,
|
|
bool alreadyClaimed,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var startedTimestamp = Stopwatch.GetTimestamp();
|
|
if ((!alreadyClaimed && job.Status != BackgroundJobStatus.Pending) ||
|
|
(alreadyClaimed && (job.Status != BackgroundJobStatus.Processing || job.LockedBy != workerId)))
|
|
{
|
|
return false;
|
|
}
|
|
await dbContext.Entry(job).ReloadAsync(cancellationToken);
|
|
if (job.CancellationRequestedAt.HasValue)
|
|
{
|
|
job.CompletedAt = DateTimeOffset.UtcNow;
|
|
await CompleteAsync(
|
|
job,
|
|
workerId,
|
|
BackgroundJobStatus.Cancelled,
|
|
job.Result,
|
|
null,
|
|
cancellationToken);
|
|
return true;
|
|
}
|
|
if (job.JobType is not ("asset_security_scan" or "tenant_export") && !(await featureAccessService.EvaluateAsync(
|
|
job.TenantId,
|
|
ResolveRequiredFeature(job.JobType, job.Payload),
|
|
FeatureAccessOperation.Write,
|
|
cancellationToken)).Allowed)
|
|
{
|
|
await CompleteAsync(
|
|
job,
|
|
workerId,
|
|
BackgroundJobStatus.Failed,
|
|
JsonDefaults.Object(),
|
|
"Tenant feature entitlement was revoked before job execution.",
|
|
cancellationToken);
|
|
return true;
|
|
}
|
|
|
|
if (!alreadyClaimed)
|
|
{
|
|
var now = DateTimeOffset.UtcNow;
|
|
job.Status = BackgroundJobStatus.Processing;
|
|
job.LockedBy = workerId;
|
|
job.LockExpiresAt = now.Add(LeaseDuration);
|
|
job.StartedAt = now;
|
|
await dbContext.SaveChangesAsync(cancellationToken);
|
|
}
|
|
|
|
try
|
|
{
|
|
var result = await tenantExecutionScope.ExecuteAsync(
|
|
new SystemScopeRequest(
|
|
job.TenantId, SystemScopeCallerType.Worker, workerId,
|
|
$"Background job {job.JobType}", job.Id.ToString("N")),
|
|
(provider, token) => ProcessCoreAsync(provider, job, token),
|
|
cancellationToken);
|
|
var cancellationRequested = await dbContext.BackgroundJobs.AsNoTracking()
|
|
.Where(item => item.Id == job.Id)
|
|
.Select(item => item.CancellationRequestedAt != null)
|
|
.SingleAsync(cancellationToken);
|
|
job.Status = cancellationRequested ? BackgroundJobStatus.Cancelled : BackgroundJobStatus.Succeeded;
|
|
job.CompletedAt = DateTimeOffset.UtcNow;
|
|
job.LastError = null;
|
|
job.Result = result;
|
|
}
|
|
catch (Exception exception) when (exception is not OperationCanceledException)
|
|
{
|
|
job.RetryCount++;
|
|
job.LastError = exception is AssetSecurityScannerException scannerException
|
|
? $"{scannerException.Code}: {scannerException.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;
|
|
}
|
|
finally
|
|
{
|
|
await CompleteAsync(job, workerId, job.Status, job.Result, job.LastError, cancellationToken);
|
|
WorkerTelemetry.RecordJob(job.JobType, job.Status.ToString(), Stopwatch.GetElapsedTime(startedTimestamp).TotalMilliseconds);
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
private async Task CompleteAsync(
|
|
BackgroundJob job,
|
|
string workerId,
|
|
BackgroundJobStatus status,
|
|
JsonElement result,
|
|
string? lastError,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
await dbContext.BackgroundJobs
|
|
.Where(value => value.Id == job.Id && value.LockedBy == workerId)
|
|
.ExecuteUpdateAsync(setters => setters
|
|
.SetProperty(value => value.Status, status)
|
|
.SetProperty(value => value.RetryCount, job.RetryCount)
|
|
.SetProperty(value => value.RunAfter, job.RunAfter)
|
|
.SetProperty(value => value.CompletedAt, job.CompletedAt)
|
|
.SetProperty(value => value.LastError, lastError)
|
|
.SetProperty(value => value.OutputAssetId, job.OutputAssetId)
|
|
.SetProperty(value => value.Result, result)
|
|
.SetProperty(value => value.LockedBy, (string?)null)
|
|
.SetProperty(value => value.LockExpiresAt, (DateTimeOffset?)null),
|
|
cancellationToken);
|
|
}
|
|
|
|
public async Task<IReadOnlyCollection<BackgroundJobItem>> ListAsync(
|
|
Guid tenantId,
|
|
string? jobType = null,
|
|
int limit = 50,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
var query = dbContext.BackgroundJobs.AsNoTracking()
|
|
.Where(job => job.TenantId == tenantId);
|
|
if (!string.IsNullOrWhiteSpace(jobType))
|
|
{
|
|
var normalized = NormalizeJobType(jobType);
|
|
query = query.Where(job => job.JobType == normalized);
|
|
}
|
|
|
|
var jobs = await query
|
|
.OrderByDescending(job => job.CreatedAt)
|
|
.Take(Math.Clamp(limit, 1, 200))
|
|
.ToArrayAsync(cancellationToken);
|
|
return jobs.Select(ToItem).ToArray();
|
|
}
|
|
|
|
public async Task<BackgroundJobItem?> GetAsync(
|
|
Guid jobId,
|
|
Guid? tenantId,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
var query = dbContext.BackgroundJobs.AsNoTracking().Where(item => item.Id == jobId);
|
|
if (tenantId.HasValue)
|
|
{
|
|
query = query.Where(item => item.TenantId == tenantId.Value);
|
|
}
|
|
var job = await query.SingleOrDefaultAsync(cancellationToken);
|
|
return job is null ? null : ToItem(job);
|
|
}
|
|
|
|
public async Task<IReadOnlyCollection<BackgroundJobItem>> ListPlatformAsync(
|
|
Guid? tenantId = null,
|
|
string? jobType = null,
|
|
BackgroundJobStatus? status = null,
|
|
int limit = 100,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
var query = dbContext.BackgroundJobs.AsNoTracking().AsQueryable();
|
|
if (tenantId.HasValue) query = query.Where(item => item.TenantId == tenantId.Value);
|
|
if (!string.IsNullOrWhiteSpace(jobType))
|
|
{
|
|
var normalized = NormalizeJobType(jobType);
|
|
query = query.Where(item => item.JobType == normalized);
|
|
}
|
|
if (status.HasValue) query = query.Where(item => item.Status == status.Value);
|
|
return (await query.OrderByDescending(item => item.CreatedAt)
|
|
.Take(Math.Clamp(limit, 1, 500))
|
|
.ToArrayAsync(cancellationToken))
|
|
.Select(ToItem)
|
|
.ToArray();
|
|
}
|
|
|
|
public async Task<BackgroundJobItem> RequestCancellationAsync(
|
|
Guid jobId,
|
|
Guid? tenantId,
|
|
Guid actorUserId,
|
|
string reason,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
dbContext.ChangeTracker.Clear();
|
|
if (string.IsNullOrWhiteSpace(reason))
|
|
{
|
|
throw new BackgroundJobException("background_job_cancel_reason_required", "Cancellation reason is required.");
|
|
}
|
|
var job = await FindMutableAsync(jobId, tenantId, cancellationToken);
|
|
if (job.Status is BackgroundJobStatus.Succeeded or BackgroundJobStatus.Failed or BackgroundJobStatus.Cancelled)
|
|
{
|
|
throw new BackgroundJobException("background_job_not_cancellable", "Only pending or processing jobs can be cancelled.");
|
|
}
|
|
|
|
var now = DateTimeOffset.UtcNow;
|
|
job.CancellationRequestedAt = now;
|
|
job.CancellationRequestedBy = actorUserId;
|
|
job.CancellationReason = reason.Trim();
|
|
if (job.Status == BackgroundJobStatus.Pending)
|
|
{
|
|
job.Status = BackgroundJobStatus.Cancelled;
|
|
job.CompletedAt = now;
|
|
}
|
|
AddMutationAudit(job, actorUserId, "background_job.cancel_requested");
|
|
await dbContext.SaveChangesAsync(cancellationToken);
|
|
return ToItem(job);
|
|
}
|
|
|
|
public async Task<BackgroundJobItem> RetryAsync(
|
|
Guid jobId,
|
|
Guid? tenantId,
|
|
Guid actorUserId,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
dbContext.ChangeTracker.Clear();
|
|
var job = await FindMutableAsync(jobId, tenantId, cancellationToken);
|
|
if (job.Status is not (BackgroundJobStatus.Failed or BackgroundJobStatus.Cancelled))
|
|
{
|
|
throw new BackgroundJobException("background_job_not_retryable", "Only failed or cancelled jobs can be retried.");
|
|
}
|
|
|
|
job.Status = BackgroundJobStatus.Pending;
|
|
job.RunAfter = DateTimeOffset.UtcNow;
|
|
job.StartedAt = null;
|
|
job.CompletedAt = null;
|
|
job.LockedBy = null;
|
|
job.LockExpiresAt = null;
|
|
job.LastError = null;
|
|
job.CancellationRequestedAt = null;
|
|
job.CancellationRequestedBy = null;
|
|
job.CancellationReason = null;
|
|
AddMutationAudit(job, actorUserId, "background_job.retry_requested");
|
|
await dbContext.SaveChangesAsync(cancellationToken);
|
|
return ToItem(job);
|
|
}
|
|
|
|
private async Task<BackgroundJob> FindMutableAsync(
|
|
Guid jobId,
|
|
Guid? tenantId,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var query = dbContext.BackgroundJobs.Where(item => item.Id == jobId);
|
|
if (tenantId.HasValue) query = query.Where(item => item.TenantId == tenantId.Value);
|
|
return await query.SingleOrDefaultAsync(cancellationToken) ??
|
|
throw new BackgroundJobException("background_job_not_found", "Background job was not found.");
|
|
}
|
|
|
|
private void AddMutationAudit(BackgroundJob job, Guid actorUserId, string action)
|
|
{
|
|
dbContext.AuditLogs.Add(new AuditLog
|
|
{
|
|
TenantId = job.TenantId,
|
|
ActorUserId = actorUserId,
|
|
Action = action,
|
|
TargetType = "background_job",
|
|
TargetId = job.Id.ToString(),
|
|
Details = JsonSerializer.SerializeToElement(new { job.JobType, job.Status })
|
|
});
|
|
}
|
|
|
|
private async Task<JsonElement> ProcessCoreAsync(
|
|
IServiceProvider scopedProvider,
|
|
BackgroundJob job,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
var scopedDbContext = scopedProvider.GetRequiredService<TikuDbContext>();
|
|
return job.JobType switch
|
|
{
|
|
"content_export" => await ProcessContentExportAsync(scopedDbContext, job, cancellationToken),
|
|
"content_import" => await ProcessContentImportAsync(scopedProvider, job, cancellationToken),
|
|
"asset_security_scan" => await ProcessAssetSecurityScanAsync(scopedProvider, scopedDbContext, job, cancellationToken),
|
|
"tenant_export" => await ProcessTenantExportAsync(scopedProvider, scopedDbContext, job, cancellationToken),
|
|
"statistics_aggregation" => await ProcessStatisticsAggregationAsync(scopedProvider, scopedDbContext, job, cancellationToken),
|
|
"commerce_reconciliation" => await ProcessCommerceReconciliationAsync(scopedDbContext, job, cancellationToken),
|
|
"tenant_domain_recheck" => await ProcessTenantDomainRecheckAsync(scopedProvider, cancellationToken),
|
|
_ => throw new InvalidOperationException($"Unsupported background job type '{job.JobType}'.")
|
|
};
|
|
}
|
|
|
|
private static async Task<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<Tiku.Application.Storage.IObjectStorageService>();
|
|
await using var content = await storage.OpenReadAsync(
|
|
new Tiku.Application.Storage.ObjectStorageReadRequest(
|
|
job.TenantId,
|
|
asset.StorageProvider switch
|
|
{
|
|
AssetStorageProvider.AliyunOss => Tiku.Application.Storage.ObjectStorageProviders.AliyunOss,
|
|
AssetStorageProvider.LocalDev => Tiku.Application.Storage.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<Tiku.Application.Storage.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, leaveOpen: false))
|
|
await using (var archive = new TarWriter(gzip, TarEntryFormat.Pax, leaveOpen: 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 => Tiku.Application.Storage.ObjectStorageProviders.AliyunOss,
|
|
AssetStorageProvider.LocalDev => Tiku.Application.Storage.ObjectStorageProviders.LocalDev,
|
|
_ => null
|
|
};
|
|
if (provider is null) continue;
|
|
await using var content = await storage.OpenReadAsync(
|
|
new Tiku.Application.Storage.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 Tiku.Application.Storage.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
|
|
{
|
|
Tiku.Application.Storage.ObjectStorageProviders.AliyunOss => AssetStorageProvider.AliyunOss,
|
|
Tiku.Application.Storage.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 System.Text.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 == Tiku.Domain.Tenancy.TenantExternalProviderCapability.Payment &&
|
|
item.Provider == provider &&
|
|
item.Status == Tiku.Domain.Tenancy.TenantExternalProviderStatus.Active,
|
|
cancellationToken);
|
|
if (!hasProviderConfig)
|
|
{
|
|
throw new InvalidOperationException($"Active payment provider '{provider}' is required for commerce reconciliation job.");
|
|
}
|
|
|
|
var billDate = GetJsonDateOnly(job.Payload, "billDate") ?? DateOnly.FromDateTime(DateTime.UtcNow.Date);
|
|
var billType = GetJsonEnum(job.Payload, "billType", ReconciliationBillType.Combined);
|
|
var sourceHash = $"background-job:{job.Id:N}";
|
|
var batch = await scopedDbContext.CommerceReconciliationBatches.SingleOrDefaultAsync(
|
|
item =>
|
|
item.TenantId == job.TenantId &&
|
|
item.Provider == provider &&
|
|
item.Source == ReconciliationSource.ProviderDownload &&
|
|
item.SourceHash == sourceHash,
|
|
cancellationToken);
|
|
if (batch is null)
|
|
{
|
|
batch = new CommerceReconciliationBatch
|
|
{
|
|
TenantId = job.TenantId,
|
|
Provider = provider,
|
|
BillDate = billDate,
|
|
BillType = billType,
|
|
Source = ReconciliationSource.ProviderDownload,
|
|
SourceName = $"provider-bill:{provider}:{billDate:yyyyMMdd}",
|
|
SourceHash = sourceHash,
|
|
Status = ReconciliationBatchStatus.Pending,
|
|
Metadata = JsonSerializer.SerializeToElement(new
|
|
{
|
|
jobId = job.Id,
|
|
note = "Provider bill job created the reconciliation batch; provider download/parser is handled by a dedicated provider processor."
|
|
})
|
|
};
|
|
scopedDbContext.CommerceReconciliationBatches.Add(batch);
|
|
await scopedDbContext.SaveChangesAsync(cancellationToken);
|
|
}
|
|
|
|
return JsonSerializer.SerializeToElement(new
|
|
{
|
|
batchId = batch.Id,
|
|
provider,
|
|
billDate,
|
|
billType = billType.ToString(),
|
|
status = batch.Status.ToString()
|
|
});
|
|
}
|
|
|
|
private static async Task<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
|
|
});
|
|
}
|
|
|
|
private static string NormalizeJobType(string jobType)
|
|
{
|
|
return jobType.Trim().ToLowerInvariant();
|
|
}
|
|
|
|
private static string? NormalizeIdempotencyKey(string? value)
|
|
{
|
|
var normalized = value?.Trim();
|
|
if (string.IsNullOrEmpty(normalized)) return null;
|
|
if (normalized.Length > 200)
|
|
{
|
|
throw new BackgroundJobException("background_job_idempotency_key_too_long", "Idempotency key cannot exceed 200 characters.");
|
|
}
|
|
return normalized;
|
|
}
|
|
|
|
private static string ResolveRequiredFeature(string jobType, JsonElement payload) => jobType switch
|
|
{
|
|
"content_import" => SaasFeatureCatalog.ResolveContentImportFeature(GetJsonString(payload, "importType"))
|
|
?? throw new InvalidOperationException("Background content import type is not supported."),
|
|
"content_export" or "asset_security_scan" => SaasFeatureCatalog.PrivateQuestionBank,
|
|
"commerce_reconciliation" => SaasFeatureCatalog.StudentStore,
|
|
"statistics_aggregation" or "tenant_domain_recheck" => SaasFeatureCatalog.CoreBackoffice,
|
|
_ => SaasFeatureCatalog.CoreBackoffice
|
|
};
|
|
|
|
private static string NormalizeProvider(string? provider)
|
|
{
|
|
var normalized = (provider ?? string.Empty).Trim().ToLowerInvariant();
|
|
return string.IsNullOrWhiteSpace(normalized)
|
|
? throw new InvalidOperationException("Background job provider is required.")
|
|
: normalized;
|
|
}
|
|
|
|
private static string? GetJsonString(JsonElement element, string propertyName)
|
|
{
|
|
return element.ValueKind == JsonValueKind.Object &&
|
|
element.TryGetProperty(propertyName, out var property) &&
|
|
property.ValueKind == JsonValueKind.String
|
|
? property.GetString()
|
|
: null;
|
|
}
|
|
|
|
private static Guid? GetJsonGuid(JsonElement element, string propertyName)
|
|
{
|
|
if (element.ValueKind != JsonValueKind.Object ||
|
|
!element.TryGetProperty(propertyName, out var property))
|
|
{
|
|
return null;
|
|
}
|
|
|
|
return property.ValueKind == JsonValueKind.String && Guid.TryParse(property.GetString(), out var value)
|
|
? value
|
|
: null;
|
|
}
|
|
|
|
private static IReadOnlyCollection<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(
|
|
job.Id,
|
|
job.TenantId,
|
|
job.JobType,
|
|
job.IdempotencyKey,
|
|
job.Status,
|
|
job.RetryCount,
|
|
job.MaxRetries,
|
|
job.RunAfter,
|
|
job.StartedAt,
|
|
job.CompletedAt,
|
|
job.CancellationRequestedAt,
|
|
job.CancellationRequestedBy,
|
|
job.CancellationReason,
|
|
job.LastError,
|
|
job.OutputAssetId,
|
|
job.Result);
|
|
}
|
|
}
|