清理代码

Tiku.Infrastructure/Jobs/Processor/BackgroundJobService.Processor.cs

@@ -1,20 +1,11 @@
-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;
+using System.Text.Json;
+using Microsoft.EntityFrameworkCore;
+using Tiku.Application.Assets;
+using Tiku.Application.Security;
+using Tiku.Domain.Common;
+using Tiku.Domain.Operations;
+using Tiku.Infrastructure.Observability;
 
 namespace Tiku.Infrastructure.Jobs;
 
@@ -29,26 +20,26 @@ internal sealed partial class BackgroundJobService
         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"
-                """)
+                                                                  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;
@@ -57,7 +48,7 @@ internal sealed partial class BackgroundJobService
         {
             cancellationToken.ThrowIfCancellationRequested();
             var job = await dbContext.BackgroundJobs.SingleAsync(value => value.Id == jobId, cancellationToken);
-            if (await ProcessJobAsync(job, workerId, alreadyClaimed: true, cancellationToken)) processed++;
+            if (await ProcessJobAsync(job, workerId, true, cancellationToken)) processed++;
             dbContext.ChangeTracker.Clear();
         }
 
@@ -81,24 +72,17 @@ internal sealed partial class BackgroundJobService
                 .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;
-        }
+        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);
+        return await ProcessJobAsync(job, workerId, true, cancellationToken);
     }
 
     private async Task<bool> ProcessJobAsync(
@@ -110,9 +94,7 @@ internal sealed partial class BackgroundJobService
         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)
         {
@@ -126,6 +108,7 @@ internal sealed partial class BackgroundJobService
                 cancellationToken);
             return true;
         }
+
         if (job.JobType is not ("asset_security_scan" or "tenant_export") && !(await featureAccessService.EvaluateAsync(
                 job.TenantId,
                 ResolveRequiredFeature(job.JobType, job.Payload),
@@ -176,9 +159,7 @@ internal sealed partial class BackgroundJobService
                 ? $"{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;
@@ -189,7 +170,8 @@ internal sealed partial class BackgroundJobService
         finally
         {
             await CompleteAsync(job, workerId, job.Status, job.Result, job.LastError, cancellationToken);
-            WorkerTelemetry.RecordJob(job.JobType, job.Status.ToString(), Stopwatch.GetElapsedTime(startedTimestamp).TotalMilliseconds);
+            WorkerTelemetry.RecordJob(job.JobType, job.Status.ToString(),
+                Stopwatch.GetElapsedTime(startedTimestamp).TotalMilliseconds);
         }
 
         return true;
@@ -206,17 +188,15 @@ internal sealed partial class BackgroundJobService
         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),
+                    .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);
     }
-
-
-}
+}