using Admin.NET.Plugin.AiDOP.DataPlatform.Executors; using Admin.NET.Plugin.AiDOP.Entity.DataPlatform; namespace Admin.NET.Plugin.AiDOP.DataPlatform; /// /// WP10 S4a · Outbox 可观测与手工重推(本库运维数据,需登录;不加 AllowAnonymous)。 /// [ApiDescriptionSettings(Order = 328, Description = "出站回写队列")] [Route("api/aidop/mdp-outbox")] [NonUnify] public class MdpOutboxAdminService : IDynamicApiController, ITransient { private readonly ISqlSugarClient _db; private readonly MdpOutboxWakeSignal _wake; public MdpOutboxAdminService(ISqlSugarClient db, MdpOutboxWakeSignal wake) { _db = db; _wake = wake; } public sealed class PageInput { public int? Status { get; set; } public string? TargetSourceCode { get; set; } public string? ActionCode { get; set; } public string? IdemKey { get; set; } public DateTime? CreateTimeFrom { get; set; } public DateTime? CreateTimeTo { get; set; } public int Page { get; set; } = 1; public int PageSize { get; set; } = 20; } public sealed class RetryInput { public long[]? Ids { get; set; } public bool? AllDead { get; set; } } [DisplayName("出站回写队列分页")] [HttpGet("page")] public async Task Page([FromQuery] PageInput input) { var page = input.Page <= 0 ? 1 : input.Page; var pageSize = input.PageSize <= 0 ? 20 : Math.Min(input.PageSize, 200); var q = _db.Queryable() .WhereIF(input.Status is 0 or 1 or 2, x => x.Status == input.Status!.Value) .WhereIF(!string.IsNullOrWhiteSpace(input.TargetSourceCode), x => x.TargetSourceCode == input.TargetSourceCode!.Trim()) .WhereIF(!string.IsNullOrWhiteSpace(input.ActionCode), x => x.ActionCode == input.ActionCode!.Trim()) .WhereIF(!string.IsNullOrWhiteSpace(input.IdemKey), x => x.IdemKey.Contains(input.IdemKey!.Trim())) .WhereIF(input.CreateTimeFrom.HasValue, x => x.CreateTime >= input.CreateTimeFrom!.Value) .WhereIF(input.CreateTimeTo.HasValue, x => x.CreateTime <= input.CreateTimeTo!.Value); RefAsync total = 0; var rows = await q.OrderByDescending(x => x.Id) .ToPageListAsync(page, pageSize, total); return new { total = total.Value, page, pageSize, list = rows.Select(x => new { id = x.Id, tenantId = x.TenantId, targetSourceCode = x.TargetSourceCode, actionCode = x.ActionCode, idemKey = x.IdemKey, payloadJson = x.PayloadJson, status = x.Status, retryCount = x.RetryCount, nextRetryTime = x.NextRetryTime?.ToString("yyyy-MM-dd HH:mm:ss"), lastErrorCode = x.LastErrorCode, responseJson = x.ResponseJson, errorMsg = x.ErrorMsg, createTime = x.CreateTime.ToString("yyyy-MM-dd HH:mm:ss"), updateTime = x.UpdateTime.ToString("yyyy-MM-dd HH:mm:ss"), }) }; } [DisplayName("出站回写队列统计")] [HttpGet("stats")] public async Task Stats() { var pending = await _db.Queryable().CountAsync(x => x.Status == 0); var success = await _db.Queryable().CountAsync(x => x.Status == 1); var dead = await _db.Queryable().CountAsync(x => x.Status == 2); var since = DateTime.Now.AddHours(-24); var deadLast24h = await _db.Queryable() .CountAsync(x => x.Status == 2 && x.UpdateTime >= since); double oldestPendingMinutes = 0; if (pending > 0) { var oldest = await _db.Queryable() .Where(x => x.Status == 0) .OrderBy(x => x.CreateTime) .Select(x => x.CreateTime) .FirstAsync(); oldestPendingMinutes = Math.Max(0, (DateTime.Now - oldest).TotalMinutes); oldestPendingMinutes = Math.Round(oldestPendingMinutes, 1); } return new { pending, success, dead, oldestPendingMinutes, deadLast24h, }; } [DisplayName("出站回写手工重推")] [HttpPost("retry")] public async Task Retry([FromBody] RetryInput input, CancellationToken ct = default) { if (input == null) throw Oops.Oh("body 不能为空"); var hasIds = input.Ids is { Length: > 0 }; var allDead = input.AllDead == true; if (!hasIds && !allDead) throw Oops.Oh("请指定 ids 或 allDead=true,禁止全表重置"); int reset; if (allDead) { reset = await _db.Updateable() .SetColumns(x => x.Status == 0) .SetColumns(x => x.RetryCount == 0) .SetColumns(x => x.ErrorMsg == null) .SetColumns(x => x.NextRetryTime == null) .SetColumns(x => x.LastErrorCode == null) .SetColumns(x => x.UpdateTime == DateTime.Now) .Where(x => x.Status == 2) .ExecuteCommandAsync(ct); } else { var ids = input.Ids!.Distinct().ToArray(); reset = await _db.Updateable() .SetColumns(x => x.Status == 0) .SetColumns(x => x.RetryCount == 0) .SetColumns(x => x.ErrorMsg == null) .SetColumns(x => x.NextRetryTime == null) .SetColumns(x => x.LastErrorCode == null) .SetColumns(x => x.UpdateTime == DateTime.Now) .Where(x => ids.Contains(x.Id)) .ExecuteCommandAsync(ct); } if (reset > 0) _wake.Pulse(); return new { reset }; } }