zengzy / hhrcs_service · Files

using System.Text.Json;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Rcs.Application.Services;
using Rcs.Application.Services.Protocol;
using Rcs.Domain.Entities;
using Rcs.Domain.Settings;
using StackExchange.Redis;

namespace Rcs.Infrastructure.Services.Protocol;

/// <summary>
/// 同车体 VDA 订单门闩与会话协调服务。
/// </summary>
public sealed class VdaOrderGateService : IVdaOrderGateService
{
    private const string GateKeyPrefix = "rcs:vda:order-gate";
    private const string SessionKeyPrefix = "rcs:vda:order-session";
    private const string NormalPrecheckFailKeyPrefix = "rcs:vda:order-precheck-fail";
    private const string ReleaseScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
    private const string RenewScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('pexpire', KEYS[1], ARGV[2]) else return 0 end";

    private readonly IConnectionMultiplexer _redis;
    private readonly IRobotCacheService _robotCacheService;
    private readonly IOptionsMonitor<AppSettings> _settingsMonitor;
    private readonly ILogger<VdaOrderGateService> _logger;

    private long _gateConflictTotal;
    private long _sessionSwitchBlockTotal;
    private long _precheckBlockTotal;
    private long _downgradeTotal;
    private long _recoveryResetTotal;

    public VdaOrderGateService(
        IConnectionMultiplexer redis,
        IRobotCacheService robotCacheService,
        IOptionsMonitor<AppSettings> settingsMonitor,
        ILogger<VdaOrderGateService> logger)
    {
        _redis = redis;
        _robotCacheService = robotCacheService;
        _settingsMonitor = settingsMonitor;
        _logger = logger;
    }

    public async Task<VdaOrderGateAcquireResult> TryAcquireAsync(
        Guid robotId,
        VdaOrderChannel channel,
        string orderId,
        int orderUpdateId,
        CancellationToken ct = default)
    {
        ct.ThrowIfCancellationRequested();
        var settings = GetVdaSettings();
        var gateKey = BuildGateKey(robotId);
        var currentSession = await GetSessionAsync(robotId, ct);

        if (channel == VdaOrderChannel.Normal && IsYieldSessionBlocking(currentSession))
        {
            IncrementMetric(
                ref _sessionSwitchBlockTotal,
                "vda_order_session_switch_block_total",
                robotId,
                channel,
                orderId,
                orderUpdateId,
                gateKey,
                currentSession?.Status,
                "yield_active");

            return new VdaOrderGateAcquireResult
            {
                Success = false,
                Message = "等待避让完成",
                GateKey = gateKey,
                CurrentSession = currentSession
            };
        }

        IVdaOrderGateLease lease;
        if (settings.EnableOrderGate)
        {
            var token = Guid.NewGuid().ToString("N");
            var ttlSeconds = Math.Max(2, settings.OrderGateTtlSeconds);
            var acquired = await _redis.GetDatabase().StringSetAsync(
                gateKey,
                token,
                TimeSpan.FromSeconds(ttlSeconds),
                When.NotExists);
            if (!acquired)
            {
                IncrementMetric(
                    ref _gateConflictTotal,
                    "vda_order_gate_conflict_total",
                    robotId,
                    channel,
                    orderId,
                    orderUpdateId,
                    gateKey,
                    currentSession?.Status,
                    "gate_conflict");

                return new VdaOrderGateAcquireResult
                {
                    Success = false,
                    Message = "订单门闩冲突，等待重试",
                    GateKey = gateKey,
                    CurrentSession = currentSession
                };
            }

            lease = new RedisVdaOrderGateLease(
                this,
                robotId,
                gateKey,
                token,
                TimeSpan.FromSeconds(ttlSeconds),
                TimeSpan.FromSeconds(Math.Max(1, settings.OrderGateRenewIntervalSeconds)),
                _logger);
        }
        else
        {
            lease = new NoopVdaOrderGateLease(robotId, gateKey);
        }

        await MarkSessionSendingAsync(robotId, channel, orderId, orderUpdateId, null, ct);
        return new VdaOrderGateAcquireResult
        {
            Success = true,
            Message = "ok",
            GateKey = gateKey,
            Lease = lease,
            CurrentSession = currentSession
        };
    }

    public Task MarkSessionSendingAsync(
        Guid robotId,
        VdaOrderChannel channel,
        string orderId,
        int orderUpdateId,
        string? precheckReason = null,
        CancellationToken ct = default)
    {
        return WriteSessionAsync(
            robotId,
            channel,
            orderId,
            orderUpdateId,
            VdaOrderSessionStatus.Sending,
            precheckReason,
            ct);
    }

    public Task MarkSessionSentAsync(
        Guid robotId,
        VdaOrderChannel channel,
        string orderId,
        int orderUpdateId,
        CancellationToken ct = default)
    {
        return WriteSessionAsync(
            robotId,
            channel,
            orderId,
            orderUpdateId,
            VdaOrderSessionStatus.Sent,
            null,
            ct);
    }

    public Task MarkSessionFailedAsync(
        Guid robotId,
        VdaOrderChannel channel,
        string orderId,
        int orderUpdateId,
        string? precheckReason = null,
        CancellationToken ct = default)
    {
        return WriteSessionAsync(
            robotId,
            channel,
            orderId,
            orderUpdateId,
            VdaOrderSessionStatus.Failed,
            precheckReason,
            ct);
    }

    public Task MarkSessionCompletedAsync(
        Guid robotId,
        VdaOrderChannel channel,
        string orderId,
        int orderUpdateId,
        CancellationToken ct = default)
    {
        return WriteSessionAsync(
            robotId,
            channel,
            orderId,
            orderUpdateId,
            VdaOrderSessionStatus.Completed,
            null,
            ct);
    }

    public async Task<VdaOrderGateSession?> GetSessionAsync(Guid robotId, CancellationToken ct = default)
    {
        ct.ThrowIfCancellationRequested();
        var payload = await _redis.GetDatabase().StringGetAsync(BuildSessionKey(robotId));
        if (payload.IsNullOrEmpty)
        {
            return null;
        }

        try
        {
            return JsonSerializer.Deserialize<VdaOrderGateSession>(payload.ToString());
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "VDA订单门闩会话解析失败: RobotId={RobotId}", robotId);
            return null;
        }
    }

    public async Task<bool> HasActiveYieldSessionAsync(
        Guid robotId,
        string? currentOrderId,
        uint? currentOrderUpdateId,
        string? lastNodeCode,
        CancellationToken ct = default)
    {
        var session = await GetSessionAsync(robotId, ct);
        if (session == null || session.Channel != VdaOrderChannel.Yield)
        {
            return false;
        }

        if (session.Status is VdaOrderSessionStatus.Failed or VdaOrderSessionStatus.Completed)
        {
            return false;
        }

        if (session.Status == VdaOrderSessionStatus.Sending)
        {
            return true;
        }

        if (string.Equals(session.OrderId, currentOrderId, StringComparison.OrdinalIgnoreCase))
        {
            if (currentOrderUpdateId.HasValue && currentOrderUpdateId.Value < (uint)Math.Max(0, session.OrderUpdateId))
            {
                return true;
            }

            return true;
        }

        if (string.IsNullOrWhiteSpace(currentOrderId) && string.IsNullOrWhiteSpace(lastNodeCode))
        {
            return true;
        }

        var recentWindow = TimeSpan.FromSeconds(Math.Max(8, GetVdaSettings().OrderGateTtlSeconds * 2));
        return DateTime.UtcNow - session.UpdatedAt.ToUniversalTime() <= recentWindow;
    }

    public async Task<VdaOrderUpdatePrecheckResult> ValidateOrderUpdatePrecheckAsync(
        Guid robotId,
        VdaOrderChannel channel,
        string? manufacturer,
        string serialNumber,
        string orderId,
        int orderUpdateId,
        CancellationToken ct = default)
    {
        var settings = GetVdaSettings();
        if (!settings.EnableOrderUpdatePrecheck || orderUpdateId <= 0)
        {
            return new VdaOrderUpdatePrecheckResult { Success = true, Reason = "precheck_skipped" };
        }

        var status = await _robotCacheService.GetStatusAsync(manufacturer, serialNumber);
        if (status == null)
        {
            IncrementMetric(
                ref _precheckBlockTotal,
                "vda_orderupdate_precheck_block_total",
                robotId,
                channel,
                orderId,
                orderUpdateId,
                BuildGateKey(robotId),
                VdaOrderSessionStatus.Failed,
                "state_missing");
            return new VdaOrderUpdatePrecheckResult { Success = false, Reason = "state_missing" };
        }

        if (status.Status == RobotStatus.Error)
        {
            IncrementMetric(
                ref _precheckBlockTotal,
                "vda_orderupdate_precheck_block_total",
                robotId,
                channel,
                orderId,
                orderUpdateId,
                BuildGateKey(robotId),
                VdaOrderSessionStatus.Failed,
                "robot_error");
            return new VdaOrderUpdatePrecheckResult
            {
                Success = false,
                Reason = "robot_error",
                RobotStatus = status.Status,
                CurrentOrderId = status.CurrentOrderId,
                CurrentOrderUpdateId = status.CurrentOrderUpdateId,
                StateTimestampUtc = status.StateTimestampUtc
            };
        }

        var stateTimestampUtc = status.StateTimestampUtc?.ToUniversalTime() ?? status.UpdatedAt.ToUniversalTime();
        var freshness = DateTime.UtcNow - stateTimestampUtc;
        if (freshness > TimeSpan.FromSeconds(Math.Max(1, settings.OrderStateFreshnessSeconds)))
        {
            IncrementMetric(
                ref _precheckBlockTotal,
                "vda_orderupdate_precheck_block_total",
                robotId,
                channel,
                orderId,
                orderUpdateId,
                BuildGateKey(robotId),
                VdaOrderSessionStatus.Failed,
                "state_stale");
            return new VdaOrderUpdatePrecheckResult
            {
                Success = false,
                Reason = "state_stale",
                RobotStatus = status.Status,
                CurrentOrderId = status.CurrentOrderId,
                CurrentOrderUpdateId = status.CurrentOrderUpdateId,
                StateTimestampUtc = stateTimestampUtc
            };
        }

        if (!string.Equals(status.CurrentOrderId ?? string.Empty, orderId, StringComparison.OrdinalIgnoreCase))
        {
            IncrementMetric(
                ref _precheckBlockTotal,
                "vda_orderupdate_precheck_block_total",
                robotId,
                channel,
                orderId,
                orderUpdateId,
                BuildGateKey(robotId),
                VdaOrderSessionStatus.Failed,
                "order_id_mismatch");
            return new VdaOrderUpdatePrecheckResult
            {
                Success = false,
                Reason = "order_id_mismatch",
                RobotStatus = status.Status,
                CurrentOrderId = status.CurrentOrderId,
                CurrentOrderUpdateId = status.CurrentOrderUpdateId,
                StateTimestampUtc = stateTimestampUtc
            };
        }

        return new VdaOrderUpdatePrecheckResult
        {
            Success = true,
            Reason = "ok",
            RobotStatus = status.Status,
            CurrentOrderId = status.CurrentOrderId,
            CurrentOrderUpdateId = status.CurrentOrderUpdateId,
            StateTimestampUtc = stateTimestampUtc
        };
    }

    public async Task<int> RegisterNormalPrecheckFailureAsync(
        Guid robotId,
        string? precheckReason,
        CancellationToken ct = default)
    {
        ct.ThrowIfCancellationRequested();
        var key = $"{NormalPrecheckFailKeyPrefix}:{robotId}";
        var db = _redis.GetDatabase();
        var count = await db.StringIncrementAsync(key);
        await db.KeyExpireAsync(key, TimeSpan.FromMinutes(5));
        _logger.LogWarning(
            "VDA订单precheck失败累计: RobotId={RobotId}, FailCount={FailCount}, PrecheckReason={PrecheckReason}",
            robotId,
            (int)count,
            precheckReason ?? string.Empty);
        return (int)count;
    }

    public Task ResetNormalPrecheckFailuresAsync(Guid robotId, CancellationToken ct = default)
    {
        ct.ThrowIfCancellationRequested();
        return _redis.GetDatabase().KeyDeleteAsync($"{NormalPrecheckFailKeyPrefix}:{robotId}");
    }

    public async Task ClearSessionAsync(Guid robotId, CancellationToken ct = default)
    {
        ct.ThrowIfCancellationRequested();
        var db = _redis.GetDatabase();
        var deletedCount = await db.KeyDeleteAsync(new RedisKey[]
        {
            BuildSessionKey(robotId),
            BuildGateKey(robotId)
        });
        _logger.LogInformation(
            "VDA订单门闩会话清理完成: RobotId={RobotId}, DeletedCount={DeletedCount}",
            robotId,
            deletedCount);
    }

    public Task RecordOrderUpdateDowngradeAsync(
        Guid robotId,
        VdaOrderChannel channel,
        string orderId,
        int orderUpdateId,
        string? precheckReason,
        CancellationToken ct = default)
    {
        ct.ThrowIfCancellationRequested();
        IncrementMetric(
            ref _downgradeTotal,
            "vda_orderupdate_downgrade_total",
            robotId,
            channel,
            orderId,
            orderUpdateId,
            BuildGateKey(robotId),
            VdaOrderSessionStatus.Sending,
            precheckReason);
        return Task.CompletedTask;
    }

    public Task RecordOrderRecoveryResetAsync(
        Guid robotId,
        string orderId,
        string? precheckReason,
        CancellationToken ct = default)
    {
        ct.ThrowIfCancellationRequested();
        IncrementMetric(
            ref _recoveryResetTotal,
            "vda_order_recovery_reset_total",
            robotId,
            VdaOrderChannel.Normal,
            orderId,
            orderUpdateId: 0,
            BuildGateKey(robotId),
            VdaOrderSessionStatus.Failed,
            precheckReason);
        return Task.CompletedTask;
    }

    private async Task WriteSessionAsync(
        Guid robotId,
        VdaOrderChannel channel,
        string orderId,
        int orderUpdateId,
        VdaOrderSessionStatus status,
        string? precheckReason,
        CancellationToken ct)
    {
        ct.ThrowIfCancellationRequested();
        var session = new VdaOrderGateSession
        {
            RobotId = robotId,
            Channel = channel,
            OrderId = orderId,
            OrderUpdateId = orderUpdateId,
            Status = status,
            UpdatedAt = DateTime.UtcNow,
            PrecheckReason = precheckReason
        };
        var payload = JsonSerializer.Serialize(session);
        var sessionTtl = TimeSpan.FromSeconds(Math.Max(60, GetVdaSettings().OrderSessionTtlSeconds));
        await _redis.GetDatabase().StringSetAsync(BuildSessionKey(robotId), payload, sessionTtl);
    }

    private bool IsYieldSessionBlocking(VdaOrderGateSession? session)
    {
        return session is
        {
            Channel: VdaOrderChannel.Yield,
            Status: VdaOrderSessionStatus.Sending or VdaOrderSessionStatus.Sent
        };
    }

    private Vda5050Settings GetVdaSettings()
    {
        return _settingsMonitor.CurrentValue.Vda5050;
    }

    private static string BuildGateKey(Guid robotId) => $"{GateKeyPrefix}:{robotId}";

    private static string BuildSessionKey(Guid robotId) => $"{SessionKeyPrefix}:{robotId}";

    private async Task<bool> ReleaseGateIfOwnedAsync(string gateKey, string token)
    {
        var result = await _redis.GetDatabase().ScriptEvaluateAsync(
            ReleaseScript,
            new RedisKey[] { gateKey },
            new RedisValue[] { token });
        return !result.IsNull && (long)result > 0;
    }

    private async Task<bool> RenewGateIfOwnedAsync(string gateKey, string token, TimeSpan ttl)
    {
        var result = await _redis.GetDatabase().ScriptEvaluateAsync(
            RenewScript,
            new RedisKey[] { gateKey },
            new RedisValue[] { token, (long)ttl.TotalMilliseconds });
        return !result.IsNull && (long)result > 0;
    }

    private void IncrementMetric(
        ref long counter,
        string metricName,
        Guid robotId,
        VdaOrderChannel? channel,
        string? orderId,
        int orderUpdateId,
        string gateKey,
        VdaOrderSessionStatus? sessionStatus,
        string? precheckReason)
    {
        var value = Interlocked.Increment(ref counter);
        _logger.LogWarning(
            "VDA订单指标: Metric={Metric}, Value={Value}, RobotId={RobotId}, Channel={Channel}, OrderId={OrderId}, OrderUpdateId={OrderUpdateId}, GateKey={GateKey}, SessionStatus={SessionStatus}, PrecheckReason={PrecheckReason}",
            metricName,
            value,
            robotId,
            channel?.ToString() ?? string.Empty,
            orderId ?? string.Empty,
            orderUpdateId,
            gateKey,
            sessionStatus?.ToString() ?? string.Empty,
            precheckReason ?? string.Empty);
    }

    private sealed class RedisVdaOrderGateLease : IVdaOrderGateLease
    {
        private readonly VdaOrderGateService _owner;
        private readonly string _token;
        private readonly TimeSpan _ttl;
        private readonly TimeSpan _renewInterval;
        private readonly ILogger _logger;
        private readonly CancellationTokenSource _renewCts = new();
        private readonly Task _renewTask;
        private int _disposed;

        public RedisVdaOrderGateLease(
            VdaOrderGateService owner,
            Guid robotId,
            string gateKey,
            string token,
            TimeSpan ttl,
            TimeSpan renewInterval,
            ILogger logger)
        {
            _owner = owner;
            RobotId = robotId;
            GateKey = gateKey;
            _token = token;
            _ttl = ttl;
            _renewInterval = renewInterval;
            _logger = logger;
            _renewTask = Task.Run(RenewLoopAsync);
        }

        public Guid RobotId { get; }
        public string GateKey { get; }

        public async ValueTask DisposeAsync()
        {
            if (Interlocked.Exchange(ref _disposed, 1) != 0)
            {
                return;
            }

            _renewCts.Cancel();
            try
            {
                await _renewTask;
            }
            catch (OperationCanceledException)
            {
                // ignored
            }
            finally
            {
                _renewCts.Dispose();
            }

            try
            {
                await _owner.ReleaseGateIfOwnedAsync(GateKey, _token);
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "VDA订单门闩释放失败: GateKey={GateKey}, RobotId={RobotId}", GateKey, RobotId);
            }
        }

        private async Task RenewLoopAsync()
        {
            while (!_renewCts.IsCancellationRequested)
            {
                try
                {
                    await Task.Delay(_renewInterval, _renewCts.Token);
                    if (_renewCts.IsCancellationRequested)
                    {
                        return;
                    }

                    var renewed = await _owner.RenewGateIfOwnedAsync(GateKey, _token, _ttl);
                    if (!renewed)
                    {
                        _logger.LogWarning(
                            "VDA订单门闩续租失败: GateKey={GateKey}, RobotId={RobotId}",
                            GateKey,
                            RobotId);
                        return;
                    }
                }
                catch (OperationCanceledException)
                {
                    return;
                }
                catch (Exception ex)
                {
                    _logger.LogWarning(ex, "VDA订单门闩续租异常: GateKey={GateKey}, RobotId={RobotId}", GateKey, RobotId);
                    return;
                }
            }
        }
    }

    private sealed class NoopVdaOrderGateLease : IVdaOrderGateLease
    {
        public NoopVdaOrderGateLease(Guid robotId, string gateKey)
        {
            RobotId = robotId;
            GateKey = gateKey;
        }

        public Guid RobotId { get; }
        public string GateKey { get; }
        public ValueTask DisposeAsync() => ValueTask.CompletedTask;
    }
}