AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/Executors/MdpOutboxEnqueueService.cs

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

namespace Admin.NET.Plugin.AiDOP.DataPlatform.Executors;

/// <summary>Outbox 入队 + 立即唤醒推送。</summary>
public sealed class MdpOutboxEnqueueService : ITransient
{
    private readonly ISqlSugarClient _db;
    private readonly MdpOutboxWakeSignal _wake;

    public MdpOutboxEnqueueService(ISqlSugarClient db, MdpOutboxWakeSignal wake)
    {
        _db = db;
        _wake = wake;
    }

    /// <summary>
    /// 幂等入队：已存在 status∈(0,1) 的同 idem_key 则跳过。
    /// </summary>
    /// <param name="pulse">为 false 时不立即 Pulse（供调用方在本地事务提交后再唤醒，见 WP10 S1）。默认 true 保持既有行为。</param>
    public async Task<bool> TryEnqueueAsync(MdpOutbox item, CancellationToken ct = default, bool pulse = true)
    {
        if (item == null) throw new ArgumentNullException(nameof(item));
        if (string.IsNullOrWhiteSpace(item.IdemKey))
            throw new ArgumentException("idem_key 不能为空", nameof(item));

        var exists = await _db.Queryable<MdpOutbox>()
            .Where(x => x.IdemKey == item.IdemKey && (x.Status == 0 || x.Status == 1))
            .AnyAsync(ct);
        if (exists) return false;

        item.Status = 0;
        item.RetryCount = 0;
        item.CreateTime = DateTime.Now;
        item.UpdateTime = item.CreateTime;
        await _db.Insertable(item).ExecuteCommandAsync(ct);
        if (pulse) _wake.Pulse();
        return true;
    }

    /// <summary>
    /// 入队或刷新：同 idem_key 已存在则删除旧行、以新 id 重新入队（供工单快照重推）。
    /// 分发器按 id 升序消费，沿用旧 id 会让本轮的先后顺序被历史入队顺序倒置（P-033）。
    /// </summary>
    public async Task<bool> TryEnqueueOrRefreshAsync(MdpOutbox item, CancellationToken ct = default, bool pulse = true)
    {
        if (item == null) throw new ArgumentNullException(nameof(item));
        if (string.IsNullOrWhiteSpace(item.IdemKey))
            throw new ArgumentException("idem_key 不能为空", nameof(item));

        var now = DateTime.Now;
        item.Status = 0;
        item.RetryCount = 0;
        item.NextRetryTime = null;
        item.LastErrorCode = null;
        item.ErrorMsg = null;
        item.ResponseJson = null;
        item.CreateTime = now;
        item.UpdateTime = now;

        // uk_mdp_outbox_idem 为 idem_key 单列唯一索引，删除与插入须在同一事务内完成
        for (var attempt = 0; attempt < 2; attempt++)
        {
            item.Id = 0;
            var tran = await _db.Ado.UseTranAsync(async () =>
            {
                await _db.Ado.ExecuteCommandAsync(
                    "DELETE FROM mdp_outbox WHERE idem_key = @IdemKey",
                    new SugarParameter("@IdemKey", item.IdemKey));
                await _db.Insertable(item).ExecuteCommandAsync();
            });

            if (tran.IsSuccess)
            {
                if (pulse) _wake.Pulse();
                return true;
            }
            // 并发下另一线程可能已抢先插入同键行，重试一次
        }

        return false;
    }
}