qhmes/yy-admin-master/YY.Admin.Services/Service/EquipmentDb/EquipmentDbCommandHandler.cs

using System.Net.Http;
using System.Net.Http.Json;
using System.Text;
using System.Text.Json;
using System.Text.RegularExpressions;
using Microsoft.Data.SqlClient;
using Microsoft.Extensions.Configuration;
using Prism.Events;
using YY.Admin.Core.Events;
using YY.Admin.Core.Services;

namespace YY.Admin.Services.Service.EquipmentDb;

/// <summary>
/// 订阅 MES 下发的 EQ_* 命令，执行设备库 SQL 并回传结果。
/// </summary>
public sealed class EquipmentDbCommandHandler
{
    private static readonly JsonSerializerOptions JsonOptions = new()
    {
        PropertyNameCaseInsensitive = true,
        PropertyNamingPolicy = JsonNamingPolicy.CamelCase
    };

    private readonly IEquipmentDbConnectionFactory _connectionFactory;
    private readonly IHttpClientFactory _httpClientFactory;
    private readonly IConfiguration _configuration;
    private readonly ISignalRService _signalRService;
    private readonly HashSet<string> _processedRequestIds = new();
    private readonly object _dedupeLock = new();

    public EquipmentDbCommandHandler(
        IEventAggregator eventAggregator,
        IEquipmentDbConnectionFactory connectionFactory,
        IHttpClientFactory httpClientFactory,
        IConfiguration configuration,
        ISignalRService signalRService)
    {
        _connectionFactory = connectionFactory;
        _httpClientFactory = httpClientFactory;
        _configuration = configuration;
        _signalRService = signalRService;

        eventAggregator.GetEvent<RemoteCommandReceivedEvent>()
            .Subscribe(OnRemoteCommand, ThreadOption.BackgroundThread);
    }

    private void OnRemoteCommand(RemoteCommandPayload payload)
    {
        if (payload == null || string.IsNullOrWhiteSpace(payload.CommandJson))
        {
            return;
        }
        try
        {
            using var doc = JsonDocument.Parse(payload.CommandJson);
            if (!doc.RootElement.TryGetProperty("cmd", out var cmdEl))
            {
                return;
            }
            var cmd = cmdEl.GetString() ?? string.Empty;
            if (!cmd.StartsWith("EQ_", StringComparison.OrdinalIgnoreCase))
            {
                return;
            }
            var request = JsonSerializer.Deserialize<EquipmentDbCommandDto>(payload.CommandJson, JsonOptions);
            if (request == null)
            {
                return;
            }
            _ = Task.Run(() => HandleAsync(request, CancellationToken.None));
        }
        catch
        {
            // ignore non-EQ payloads
        }
    }

    public async Task HandleAsync(EquipmentDbCommandDto request, CancellationToken cancellationToken)
    {
        if (request == null || string.IsNullOrWhiteSpace(request.RequestId))
        {
            return;
        }
        lock (_dedupeLock)
        {
            if (!_processedRequestIds.Add(request.RequestId))
            {
                return;
            }
            if (_processedRequestIds.Count > 2000)
            {
                _processedRequestIds.Clear();
                _processedRequestIds.Add(request.RequestId);
            }
        }

        var result = new EquipmentDbResultDto
        {
            RequestId = request.RequestId,
            DeviceId = ResolveDeviceId(request.DeviceId),
            Cmd = request.Cmd,
            RespondedAt = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
        };

        try
        {
            switch ((request.Cmd ?? string.Empty).ToUpperInvariant())
            {
                case "EQ_TEST":
                    await HandleTestAsync(result, cancellationToken).ConfigureAwait(false);
                    break;
                case "EQ_QUERY":
                case "EQ_SYNC_PULL":
                    await HandleQueryAsync(request, result, cancellationToken).ConfigureAwait(false);
                    break;
                case "EQ_WRITE":
                    await HandleWriteAsync(request, result, cancellationToken).ConfigureAwait(false);
                    break;
                case "EQ_SYNC_PUSH":
                    await HandleSyncPushAsync(request, result, cancellationToken).ConfigureAwait(false);
                    break;
                default:
                    result.Success = false;
                    result.Error = "未知命令: " + request.Cmd;
                    break;
            }
        }
        catch (Exception ex)
        {
            result.Success = false;
            result.Error = ex.Message;
            result.DbConnected = false;
            result.DbMessage = ex.Message;
        }

        await SendResultAsync(result, cancellationToken).ConfigureAwait(false);
    }

    private async Task HandleTestAsync(EquipmentDbResultDto result, CancellationToken cancellationToken)
    {
        var (ok, message) = await _connectionFactory.TestAsync(cancellationToken).ConfigureAwait(false);
        result.Success = ok;
        result.DbConnected = ok;
        result.DbMessage = message;
        result.Error = ok ? null : message;
        await ReportStatusAsync(ok, message, cancellationToken).ConfigureAwait(false);
    }

    private async Task HandleQueryAsync(EquipmentDbCommandDto request, EquipmentDbResultDto result, CancellationToken cancellationToken)
    {
        EquipmentDbWhitelist.ValidateSql(request.Sql, request.Table);
        await using var conn = _connectionFactory.CreateConnection();
        await conn.OpenAsync(cancellationToken).ConfigureAwait(false);
        await using var cmd = conn.CreateCommand();
        cmd.CommandText = request.Sql!;
        cmd.CommandTimeout = 60;
        BindParams(cmd, request.Params);
        var rows = new List<Dictionary<string, object?>>();
        await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
        {
            var row = new Dictionary<string, object?>(StringComparer.OrdinalIgnoreCase);
            for (var i = 0; i < reader.FieldCount; i++)
            {
                object? val = reader.IsDBNull(i) ? null : reader.GetValue(i);
                if (val is DateTime dt)
                {
                    val = dt.ToString("yyyy-MM-dd HH:mm:ss");
                }
                row[reader.GetName(i)] = val;
            }
            rows.Add(row);
        }
        result.Success = true;
        result.Rows = rows;
        result.Affected = rows.Count;
        result.DbConnected = true;
        result.DbMessage = "ok";
    }

    private async Task HandleWriteAsync(EquipmentDbCommandDto request, EquipmentDbResultDto result, CancellationToken cancellationToken)
    {
        EquipmentDbWhitelist.ValidateSql(request.Sql, request.Table);
        await using var conn = _connectionFactory.CreateConnection();
        await conn.OpenAsync(cancellationToken).ConfigureAwait(false);
        await using var cmd = conn.CreateCommand();
        cmd.CommandText = request.Sql!;
        cmd.CommandTimeout = 60;
        BindParams(cmd, request.Params);
        var affected = await cmd.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
        result.Success = true;
        result.Affected = affected;
        result.DbConnected = true;
        result.DbMessage = "ok";
    }

    private async Task HandleSyncPushAsync(EquipmentDbCommandDto request, EquipmentDbResultDto result, CancellationToken cancellationToken)
    {
        if (string.IsNullOrWhiteSpace(request.Table) || !EquipmentDbWhitelist.IsAllowedTable(request.Table))
        {
            throw new InvalidOperationException("EQ_SYNC_PUSH 表不在白名单");
        }
        if (request.Payload == null || request.Payload.Count == 0)
        {
            throw new InvalidOperationException("EQ_SYNC_PUSH payload 为空");
        }

        request.Payload.TryGetValue("mode", out var modeObj);
        var mode = Convert.ToString(modeObj)?.ToLowerInvariant() ?? "insert";
        var rows = ExtractRows(request.Payload);
        if (rows.Count == 0)
        {
            result.Success = true;
            result.Affected = 0;
            result.DbConnected = true;
            return;
        }

        List<string>? keys = null;
        if (request.Payload.TryGetValue("keys", out var keysObj) && keysObj != null)
        {
            keys = JsonSerializer.Deserialize<List<string>>(JsonSerializer.Serialize(keysObj), JsonOptions);
        }

        await using var conn = _connectionFactory.CreateConnection();
        await conn.OpenAsync(cancellationToken).ConfigureAwait(false);
        var affected = 0;
        foreach (var row in rows)
        {
            if (mode == "upsert" && keys != null && keys.Count > 0)
            {
                affected += await UpsertRowAsync(conn, request.Table!, row, keys, cancellationToken).ConfigureAwait(false);
            }
            else if (mode == "update" && keys != null && keys.Count > 0)
            {
                affected += await UpdateRowAsync(conn, request.Table!, row, keys, cancellationToken).ConfigureAwait(false);
            }
            else
            {
                affected += await InsertRowAsync(conn, request.Table!, row, cancellationToken).ConfigureAwait(false);
            }
        }
        result.Success = true;
        result.Affected = affected;
        result.DbConnected = true;
        result.DbMessage = "ok";
    }

    private static List<Dictionary<string, object?>> ExtractRows(Dictionary<string, object?> payload)
    {
        if (!payload.TryGetValue("rows", out var rowsObj) || rowsObj == null)
        {
            throw new InvalidOperationException("EQ_SYNC_PUSH 缺少 rows");
        }
        if (rowsObj is JsonElement je)
        {
            return JsonSerializer.Deserialize<List<Dictionary<string, object?>>>(je.GetRawText(), JsonOptions)
                   ?? new List<Dictionary<string, object?>>();
        }
        return JsonSerializer.Deserialize<List<Dictionary<string, object?>>>(JsonSerializer.Serialize(rowsObj), JsonOptions)
               ?? new List<Dictionary<string, object?>>();
    }

    private static async Task<int> InsertRowAsync(
        SqlConnection conn, string table, Dictionary<string, object?> row, CancellationToken cancellationToken)
    {
        var cols = row.Keys.ToList();
        foreach (var c in cols)
        {
            if (!Regex.IsMatch(c, @"^[A-Za-z0-9_]+$"))
            {
                throw new InvalidOperationException("非法列名: " + c);
            }
        }
        var colList = string.Join(",", cols.Select(c => "[" + c + "]"));
        var paramList = string.Join(",", cols.Select((_, i) => "@p" + i));
        await using var cmd = conn.CreateCommand();
        cmd.CommandText = $"INSERT INTO [{table}] ({colList}) VALUES ({paramList})";
        for (var i = 0; i < cols.Count; i++)
        {
            cmd.Parameters.AddWithValue("@p" + i, ToDbValue(row[cols[i]]));
        }
        return await cmd.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
    }

    private static async Task<int> UpdateRowAsync(
        SqlConnection conn, string table, Dictionary<string, object?> row, List<string> keys, CancellationToken cancellationToken)
    {
        var setCols = row.Keys.Where(k => !keys.Contains(k, StringComparer.OrdinalIgnoreCase)).ToList();
        if (setCols.Count == 0)
        {
            return 0;
        }
        var sets = string.Join(",", setCols.Select((c, i) => $"[{c}]=@s{i}"));
        var wheres = string.Join(" AND ", keys.Select((k, i) => $"[{k}]=@k{i}"));
        await using var cmd = conn.CreateCommand();
        cmd.CommandText = $"UPDATE [{table}] SET {sets} WHERE {wheres}";
        for (var i = 0; i < setCols.Count; i++)
        {
            cmd.Parameters.AddWithValue("@s" + i, ToDbValue(GetIgnoreCase(row, setCols[i])));
        }
        for (var i = 0; i < keys.Count; i++)
        {
            cmd.Parameters.AddWithValue("@k" + i, ToDbValue(GetIgnoreCase(row, keys[i])));
        }
        return await cmd.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
    }

    private static async Task<int> UpsertRowAsync(
        SqlConnection conn, string table, Dictionary<string, object?> row, List<string> keys, CancellationToken cancellationToken)
    {
        var updated = await UpdateRowAsync(conn, table, row, keys, cancellationToken).ConfigureAwait(false);
        if (updated > 0)
        {
            return updated;
        }
        return await InsertRowAsync(conn, table, row, cancellationToken).ConfigureAwait(false);
    }

    private static object? GetIgnoreCase(Dictionary<string, object?> row, string key)
    {
        if (row.TryGetValue(key, out var v))
        {
            return v;
        }
        foreach (var kv in row)
        {
            if (string.Equals(kv.Key, key, StringComparison.OrdinalIgnoreCase))
            {
                return kv.Value;
            }
        }
        return null;
    }

    private static object ToDbValue(object? value)
    {
        if (value == null)
        {
            return DBNull.Value;
        }
        if (value is JsonElement je)
        {
            return je.ValueKind switch
            {
                JsonValueKind.Null => DBNull.Value,
                JsonValueKind.String => je.GetString() ?? (object)DBNull.Value,
                JsonValueKind.Number => je.TryGetInt64(out var l) ? l : je.GetDouble(),
                JsonValueKind.True => true,
                JsonValueKind.False => false,
                _ => je.ToString()
            };
        }
        return value;
    }

    private static void BindParams(SqlCommand cmd, List<object?>? parameters)
    {
        if (parameters == null)
        {
            return;
        }
        for (var i = 0; i < parameters.Count; i++)
        {
            cmd.Parameters.AddWithValue("@p" + i, ToDbValue(parameters[i]));
        }
        if (cmd.CommandText.Contains('?'))
        {
            var sb = new StringBuilder();
            var idx = 0;
            foreach (var ch in cmd.CommandText)
            {
                if (ch == '?')
                {
                    sb.Append("@p").Append(idx++);
                }
                else
                {
                    sb.Append(ch);
                }
            }
            cmd.CommandText = sb.ToString();
        }
    }

    private async Task SendResultAsync(EquipmentDbResultDto result, CancellationToken cancellationToken)
    {
        var json = JsonSerializer.Serialize(result, JsonOptions);
        var useRest = json.Length > 200_000 || (result.Rows?.Count ?? 0) > 500;
        if (useRest)
        {
            try
            {
                var client = _httpClientFactory.CreateClient("JeecgApi");
                result.LastChunk = true;
                result.ChunkIndex = 0;
                result.ChunkTotal = 1;
                var resp = await client.PostAsJsonAsync("/xslmes/deviceAgent/anon/resultChunk", result, cancellationToken)
                    .ConfigureAwait(false);
                resp.EnsureSuccessStatusCode();
                return;
            }
            catch
            {
                // fall through
            }
        }

        if (_signalRService is IEquipmentDbStompSender sender)
        {
            await sender.SendAgentResultAsync(result, cancellationToken).ConfigureAwait(false);
        }
    }

    public async Task ReportStatusAsync(bool dbConnected, string? dbMessage, CancellationToken cancellationToken = default)
    {
        var deviceId = ResolveDeviceId(null);
        var status = new Dictionary<string, object?>
        {
            ["deviceId"] = deviceId,
            ["hostName"] = Environment.MachineName,
            ["dbConnected"] = dbConnected,
            ["dbMessage"] = dbMessage,
            ["ts"] = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
        };
        if (_signalRService is IEquipmentDbStompSender sender)
        {
            await sender.SendAgentStatusAsync(status, cancellationToken).ConfigureAwait(false);
        }
        try
        {
            var client = _httpClientFactory.CreateClient("JeecgApi");
            await client.PostAsJsonAsync("/xslmes/deviceAgent/anon/heartbeat", status, cancellationToken).ConfigureAwait(false);
        }
        catch
        {
            // ignore
        }
    }

    private string ResolveDeviceId(string? fromRequest)
    {
        if (!string.IsNullOrWhiteSpace(fromRequest))
        {
            return fromRequest!;
        }
        var settings = _connectionFactory.CurrentSettings;
        if (!string.IsNullOrWhiteSpace(settings.DeviceId))
        {
            return settings.DeviceId;
        }
        return _configuration.GetValue<string>("EquipmentDb:DeviceId")
               ?? Environment.MachineName;
    }
}

public sealed class EquipmentDbCommandDto
{
    public string? RequestId { get; set; }
    public string? Cmd { get; set; }
    public string? DeviceId { get; set; }
    public string? Table { get; set; }
    public string? Sql { get; set; }
    public string? SqlKey { get; set; }
    public List<object?>? Params { get; set; }
    public Dictionary<string, object?>? Payload { get; set; }
    public int? TimeoutMs { get; set; }
}

public sealed class EquipmentDbResultDto
{
    public string? RequestId { get; set; }
    public string? DeviceId { get; set; }
    public string? Cmd { get; set; }
    public bool Success { get; set; }
    public string? Error { get; set; }
    public int? Affected { get; set; }
    public List<Dictionary<string, object?>>? Rows { get; set; }
    public bool? DbConnected { get; set; }
    public string? DbMessage { get; set; }
    public long? RespondedAt { get; set; }
    public int? ChunkIndex { get; set; }
    public int? ChunkTotal { get; set; }
    public bool? LastChunk { get; set; }
}