using Microsoft.Extensions.Logging; using Newtonsoft.Json; using QYZH.InteractiveMagazine.IService; using QYZH.InteractiveMagazine.Models.Dto.Compensation; using QYZH.InteractiveMagazine.Models.Entity; using QYZH.InteractiveMagazine.Repository; namespace QYZH.InteractiveMagazine.Service; /// /// 补偿任务服务实现(仅负责记录,处理逻辑由外部 Hangfire 项目完成) /// public class CompensationTaskService( BaseRepository taskRepository, ILogger logger) : BaseRepository, ICompensationTaskService { /// /// 创建补偿任务 — 记录失败操作,供后续补偿处理 /// public async Task CreateTaskAsync(CreateCompensationTaskInput input) { logger.LogWarning( "创建补偿任务,TaskType: {TaskType}, BusinessSource: {BusinessSource}, UserId: {UserId}, ErrorSource: {ErrorSource}, Error: {ErrorMessage}", input.TaskType, input.BusinessSource, input.UserId, input.ErrorSource, input.ErrorMessage); var payloadJson = input.Payload is string str ? str : JsonConvert.SerializeObject(input.Payload); var task = new CompensationTask { TaskType = input.TaskType, BusinessSource = input.BusinessSource, BusinessId = input.BusinessId, UserId = input.UserId, Payload = payloadJson, ErrorMessage = input.ErrorMessage, ErrorSource = input.ErrorSource, RetryCount = 0, MaxRetries = input.MaxRetries > 0 ? input.MaxRetries : 3, Status = CompensationTaskStatus.Pending, ScheduledAt = DateTime.Now, IsDeleted = false, CreatedBy = "System", CreatedAt = DateTime.Now, UpdatedBy = "System", UpdatedAt = DateTime.Now }; var result = await taskRepository.InsertReturnEntityAsync(task); logger.LogInformation("补偿任务创建成功,TaskId: {TaskId}", result.Id); return result.Id; } /// /// 获取待处理的补偿任务列表 /// public async Task> GetPendingTasksAsync(int limit = 50) { var now = DateTime.Now; var tasks = await taskRepository.Queryable() .Where(t => (t.Status == CompensationTaskStatus.Pending || t.Status == CompensationTaskStatus.Processing) && !t.IsDeleted && (t.ScheduledAt == null || t.ScheduledAt <= now)) .OrderBy(t => t.CreatedAt) .Take(limit) .Select(t => new CompensationTaskOutput { Id = t.Id, TaskType = t.TaskType, BusinessSource = t.BusinessSource, BusinessId = t.BusinessId, UserId = t.UserId, Payload = t.Payload, ErrorMessage = t.ErrorMessage, ErrorSource = t.ErrorSource, RetryCount = t.RetryCount, MaxRetries = t.MaxRetries, Status = t.Status, ProcessedAt = t.ProcessedAt, ScheduledAt = t.ScheduledAt, ResultMessage = t.ResultMessage, CreatedAt = t.CreatedAt }) .ToListAsync(); return tasks; } /// /// 根据业务来源和类型查询补偿任务(用于外部项目按条件拉取) /// public async Task> GetTasksAsync(GetCompensationTasksInput input) { var query = taskRepository.Queryable() .Where(t => !t.IsDeleted); if (!string.IsNullOrEmpty(input.Status)) query = query.Where(t => t.Status == input.Status); if (!string.IsNullOrEmpty(input.TaskType)) query = query.Where(t => t.TaskType == input.TaskType); if (!string.IsNullOrEmpty(input.BusinessSource)) query = query.Where(t => t.BusinessSource == input.BusinessSource); var tasks = await query .OrderBy(t => t.CreatedAt) .Take(input.Limit > 0 ? input.Limit : 50) .Select(t => new CompensationTaskOutput { Id = t.Id, TaskType = t.TaskType, BusinessSource = t.BusinessSource, BusinessId = t.BusinessId, UserId = t.UserId, Payload = t.Payload, ErrorMessage = t.ErrorMessage, ErrorSource = t.ErrorSource, RetryCount = t.RetryCount, MaxRetries = t.MaxRetries, Status = t.Status, ProcessedAt = t.ProcessedAt, ScheduledAt = t.ScheduledAt, ResultMessage = t.ResultMessage, CreatedAt = t.CreatedAt }) .ToListAsync(); return tasks; } /// /// 更新补偿任务状态(供外部处理项目回调更新结果) /// public async Task UpdateTaskStatusAsync(long taskId, UpdateCompensationStatusInput input) { var update = taskRepository.Context.Updateable() .SetColumns(t => t.Status == input.Status) .SetColumns(t => t.ResultMessage == input.ResultMessage) .SetColumns(t => t.ProcessedAt == DateTime.Now) .SetColumns(t => t.UpdatedAt == DateTime.Now); // 如果外部传入了重试相关字段,一并更新 if (input.RetryCount.HasValue) update = update.SetColumns(t => t.RetryCount == input.RetryCount.Value); if (input.ScheduledAt.HasValue) update = update.SetColumns(t => t.ScheduledAt == input.ScheduledAt.Value); await update.Where(t => t.Id == taskId && !t.IsDeleted) .ExecuteCommandAsync(); logger.LogInformation("补偿任务状态更新,TaskId: {TaskId}, Status: {Status}", taskId, input.Status); } /// /// 取消补偿任务 /// public async Task CancelTaskAsync(long taskId, string reason) { await taskRepository.Context.Updateable() .SetColumns(t => t.Status == CompensationTaskStatus.Cancelled) .SetColumns(t => t.ResultMessage == reason) .SetColumns(t => t.UpdatedAt == DateTime.Now) .Where(t => t.Id == taskId && !t.IsDeleted) .ExecuteCommandAsync(); logger.LogInformation("补偿任务已取消,TaskId: {TaskId}, Reason: {Reason}", taskId, reason); } }