AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/MdpOutboxAdminService.cs

using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;

namespace Admin.NET.Plugin.AiDOP.DataPlatform;

/// <summary>
/// WP10 S4a · Outbox 可观测与手工重推（本库运维数据，需登录；不加 AllowAnonymous）。
/// </summary>
[ApiDescriptionSettings(Order = 328, Description = "出站回写队列")]
[Route("api/aidop/mdp-outbox")]
[NonUnify]
public class MdpOutboxAdminService : IDynamicApiController, ITransient
{
    private readonly ISqlSugarClient _db;
    private readonly MdpOutboxWakeSignal _wake;
    private readonly MdpDbPushExecutor _dbPush;
    private readonly MdpApiPushExecutor _apiPush;

    public MdpOutboxAdminService(
        ISqlSugarClient db,
        MdpOutboxWakeSignal wake,
        MdpDbPushExecutor dbPush,
        MdpApiPushExecutor apiPush)
    {
        _db = db;
        _wake = wake;
        _dbPush = dbPush;
        _apiPush = apiPush;
    }

    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<object> 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<MdpOutbox>()
            .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<int> 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<object> Stats()
    {
        var pending = await _db.Queryable<MdpOutbox>().CountAsync(x => x.Status == 0);
        var success = await _db.Queryable<MdpOutbox>().CountAsync(x => x.Status == 1);
        var dead = await _db.Queryable<MdpOutbox>().CountAsync(x => x.Status == 2);

        var since = DateTime.Now.AddHours(-24);
        var deadLast24h = await _db.Queryable<MdpOutbox>()
            .CountAsync(x => x.Status == 2 && x.UpdateTime >= since);

        double oldestPendingMinutes = 0;
        if (pending > 0)
        {
            var oldest = await _db.Queryable<MdpOutbox>()
                .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<object> 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<MdpOutbox>()
                .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<MdpOutbox>()
                .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 };
    }

    /// <summary>
    /// 本进程内立刻执行一条 Outbox（不经集群/其它实例抢消费），用于 PROC 等新能力验收。
    /// </summary>
    [DisplayName("出站回写本机立即执行")]
    [HttpPost("push-now/{id:long}")]
    public async Task<object> PushNow(long id, CancellationToken ct = default)
    {
        var item = await _db.Queryable<MdpOutbox>().InSingleAsync(id);
        if (item == null)
            throw Oops.Oh($"outbox id={id} 不存在");

        var source = await _db.Queryable<MdpSource>()
            .Where(x => x.SourceCode == item.TargetSourceCode && x.Status == 1)
            .FirstAsync(ct);
        if (source == null)
            throw Oops.Oh($"目标源 {item.TargetSourceCode} 未启用");

        IMdpTargetPushExecutor executor = string.Equals(source.SourceType, "DB", StringComparison.OrdinalIgnoreCase)
            ? _dbPush
            : _apiPush;

        var result = await executor.PushAsync(source, item, ct);
        var now = DateTime.Now;
        if (result.Success)
        {
            await _db.Updateable<MdpOutbox>()
                .SetColumns(x => x.Status == 1)
                .SetColumns(x => x.ErrorMsg == null)
                .SetColumns(x => x.LastErrorCode == null)
                .SetColumns(x => x.NextRetryTime == null)
                .SetColumns(x => x.ResponseJson == result.ResponseJson)
                .SetColumns(x => x.UpdateTime == now)
                .Where(x => x.Id == id)
                .ExecuteCommandAsync(ct);
        }
        else
        {
            await _db.Updateable<MdpOutbox>()
                .SetColumns(x => x.Status == 2)
                .SetColumns(x => x.ErrorMsg == (result.ErrorMessage ?? "push failed"))
                .SetColumns(x => x.ResponseJson == result.ResponseJson)
                .SetColumns(x => x.NextRetryTime == null)
                .SetColumns(x => x.UpdateTime == now)
                .Where(x => x.Id == id)
                .ExecuteCommandAsync(ct);
        }

        return new
        {
            id,
            success = result.Success,
            idempotentSkip = result.IdempotentSkip,
            errorMessage = result.ErrorMessage,
            responseJson = result.ResponseJson,
            affectedRows = result.AffectedRows
        };
    }
}