41 lines
3.3 KiB
C#
41 lines
3.3 KiB
C#
using System.Data;
|
||
using IM.Admin.Data;
|
||
using IM.InitCommon.Management;
|
||
using Microsoft.EntityFrameworkCore;
|
||
|
||
namespace IM.Admin.Services;
|
||
public sealed class OperationWorker(IServiceScopeFactory scopes, ILogger<OperationWorker> logger) : BackgroundService
|
||
{
|
||
protected override async Task ExecuteAsync(CancellationToken ct)
|
||
{
|
||
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(2));
|
||
while (await timer.WaitForNextTickAsync(ct)) {
|
||
try { await Process(ct); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { break; } catch (Exception e) { logger.LogWarning("Operation dispatcher unavailable: {Type}", e.GetType().Name); }
|
||
}
|
||
}
|
||
public async Task Process(CancellationToken ct)
|
||
{
|
||
using var scope = scopes.CreateScope(); var db = scope.ServiceProvider.GetRequiredService<AdminDb>();
|
||
var id = await db.Operations.Where(x => x.Status == "pending" && x.NextAttemptAt <= DateTime.UtcNow).OrderBy(x => x.CreatedAt).Select(x => (Guid?)x.Id).FirstOrDefaultAsync(ct);
|
||
if (id is null) return;
|
||
await using var tx = await db.Database.BeginTransactionAsync(IsolationLevel.Serializable, ct);
|
||
// A row lock serializes dispatchers. Receipt idempotency covers response loss after the domain commit.
|
||
var op = await db.Operations.FromSqlInterpolated($"SELECT * FROM admin_operations WHERE Id = {id.Value} FOR UPDATE").SingleAsync(ct);
|
||
if (op.Status != "pending" || op.NextAttemptAt > DateTime.UtcNow) return;
|
||
try {
|
||
var receipt = op.Action is "警告" or "驳回" ? new ActionReceipt(op.TargetId.ToString(), "处理中", op.Action == "驳回" ? "已驳回" : "已处理") : await scope.ServiceProvider.GetRequiredService<InternalClient>().Send<ActionReceipt>(op.Type, "/internal/management/action", new InternalAction(op.Id, op.ActorId, op.TargetId, op.Action, op.Reason), ct);
|
||
if (op.ReportId.HasValue) {
|
||
var r = await db.Reports.SingleAsync(x => x.Id == op.ReportId, ct);
|
||
r.Status = op.Action == "驳回" ? "已驳回" : "已处理"; r.Result = op.Action + "：" + op.Reason; r.ClosedAt = DateTime.UtcNow; r.Version++;
|
||
}
|
||
op.Status = "completed"; op.CompletedAt = DateTime.UtcNow; op.Error = null;
|
||
db.Audit.Add(new AuditRecord { ActorId = op.ActorId, ActorName = op.ActorName, Action = op.Action, TargetId = op.TargetId.ToString(), TargetName = receipt.TargetName, Before = receipt.Before, After = receipt.After, Reason = op.Reason, ReportId = op.ReportId, OperationId = op.Id });
|
||
} catch (Exception e) when (e is not OperationCanceledException || !ct.IsCancellationRequested) {
|
||
op.Attempts++; op.Error = e is InternalServiceException se ? se.Message : "业务服务暂不可用，可重试原任务";
|
||
op.Status = op.Attempts >= 3 ? "failed" : "pending"; op.NextAttemptAt = DateTime.UtcNow.AddSeconds(10 * op.Attempts);
|
||
db.Audit.Add(new AuditRecord { ActorId = op.ActorId, ActorName = op.ActorName, Action = op.Action, TargetId = op.TargetId.ToString(), TargetName = op.TargetId.ToString(), Before = "待确认", After = "未确认", Reason = op.Reason, ReportId = op.ReportId, Result = $"第 {op.Attempts} 次执行失败；任务 {op.Id}" });
|
||
}
|
||
await db.SaveChangesAsync(ct); await tx.CommitAsync(ct);
|
||
}
|
||
}
|