TestPlatfoming.git

using System;
using System.Buffers.Binary;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data;
using System.Globalization;
using System.IO;
using System.IO.Pipes;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Linq;
using Aging.Infrastructure.Broker;
using Aging.Infrastructure.Contracts;
using Aging.Infrastructure.Core;
using Aging.Infrastructure.Drivers;
using Aging.ResourceContracts;
using log4net;
 
namespace DeviceBroker
{
    /// <summary>
    /// DeviceBroker 宿主：唯一持有 Broker 域设备物理会话的进程（V3.0 混合架构）。
    /// 出队时双重校验：arb.db 栅栏（epoch/generation/相位，INV-01/03）+ configHash 对账（INV-04），
    /// SessionGate 把“选择通道 + 写入/查询 + 读取响应”视为不可拆分原子片段串行执行（§04）。
    /// MVP 为 OneShot 命令同步模型：周期任务/订阅/Outbox 推迟（P2+）。
    /// </summary>
    public sealed class BrokerServer
    {
        private static readonly ILog Log = LogManager.GetLogger(typeof(BrokerServer));
 
        private static readonly string[] AllowedOps = { "CONNECT", "DISCONNECT", "SEND", "QUERY", "STATUS" };
 
        private readonly string _brokerId;
        private readonly string _pipeName;
        private readonly string _configDbPath;
        private readonly DbLeaseStore _lease;
 
        private sealed class ManagedDevice
        {
            public string DeviceId = "";
            public string Transport = "";
            public string Endpoint = "";
            public string SessionKey = "";
            public IDeviceDriver? Driver;
            public bool Connected;
        }
 
        private readonly Dictionary<string, ManagedDevice> _devices =
            new(StringComparer.OrdinalIgnoreCase);
        private readonly ConcurrentDictionary<string, SemaphoreSlim> _gates = new();
        private readonly SemaphoreSlim _driverCreateLock = new(1, 1);
 
        private long _hashStampMs;
        private volatile string _cachedHash = "";
 
        public BrokerServer(string configDbPath, string arbDbPath, string brokerId)
        {
            _configDbPath = configDbPath;
            _brokerId = brokerId;
            _pipeName = BrokerPipeName.For(brokerId);
            _lease = new DbLeaseStore(arbDbPath, brokerId);
            // 宿主重启对账：回收上次进程残留的待回收租约（INV-03 自动化形式，证据落库）
            _lease.ReconcileAfterBoot($"broker:{brokerId}/{System.Diagnostics.Process.GetCurrentProcess().Id}");
        }
 
        public long Epoch => _lease.CurrentEpoch;
 
        /// <summary>加载本 Broker 域设备清单（ExecutionMode=Broker 且 BrokerId 匹配）。</summary>
        public void LoadDevices()
        {
            _devices.Clear();
            using var conn = new System.Data.SQLite.SQLiteConnection(
                $"Data Source={_configDbPath};Version=3;Read Only=True;");
            conn.Open();
            using var cmd = conn.CreateCommand();
            cmd.CommandText = @"SELECT DeviceId, Transport, Endpoint, PhysicalSessionKey, BrokerId, AuthorityId
                                FROM DeviceDefinition WHERE ExecutionMode = 'Broker';";
            using var rd = cmd.ExecuteReader();
            var dt = new DataTable();
            dt.Load(rd);
            foreach (DataRow row in dt.Rows)
            {
                var brokerId = row["BrokerId"].ToString() ?? "";
                if (!string.Equals(brokerId, _brokerId, StringComparison.OrdinalIgnoreCase))
                    continue; // 属其他 Broker 域，本宿主拒绝代收
                _devices[row["DeviceId"].ToString()!] = new ManagedDevice
                {
                    DeviceId = row["DeviceId"].ToString() ?? "",
                    Transport = row["Transport"].ToString() ?? "",
                    Endpoint = row["Endpoint"].ToString() ?? "",
                    SessionKey = string.IsNullOrWhiteSpace(row["PhysicalSessionKey"].ToString())
                        ? row["DeviceId"].ToString()! // 无会话键时退化为按设备串行
                        : row["PhysicalSessionKey"].ToString()!,
                };
            }
            Log.Info($"[Broker] {_brokerId} 载入 {_devices.Count} 台 Broker 设备, epoch={_lease.CurrentEpoch}");
        }
 
        /// <summary>接受循环：每个连接一个处理任务；管道保持多实例可用。</summary>
        public async Task RunAsync(CancellationToken ct)
        {
            Log.Info($"[Broker] 监听管道 [{_pipeName}]");
            while (!ct.IsCancellationRequested)
            {
                NamedPipeServerStream? stream = null;
                try
                {
                    stream = new NamedPipeServerStream(_pipeName,
                        PipeDirection.InOut, 8, PipeTransmissionMode.Byte, PipeOptions.Asynchronous);
                    await stream.WaitForConnectionAsync(ct).ConfigureAwait(false);
                    var s = stream;
                    stream = null;
                    _ = Task.Run(() => HandleConnectionAsync(s, ct), CancellationToken.None);
                }
                catch (OperationCanceledException)
                {
                    break;
                }
                catch (Exception ex)
                {
                    Log.Error($"[Broker] 接受连接异常: {ex.Message}", ex);
                    stream?.Dispose();
                    try { await Task.Delay(500, ct).ConfigureAwait(false); }
                    catch (OperationCanceledException) { break; }
                }
            }
        }
 
        private async Task HandleConnectionAsync(NamedPipeServerStream stream, CancellationToken ct)
        {
            var lenBuf = new byte[4];
            try
            {
                using (stream)
                {
                    while (!ct.IsCancellationRequested && stream.IsConnected)
                    {
                        if (!await ReadExactAsync(stream, lenBuf, 4, ct).ConfigureAwait(false)) break;
                        int len = BinaryPrimitives.ReadInt32BigEndian(lenBuf);
                        if (len <= 0 || len > BrokerIpc.MaxFrameBytes) break;
                        var payload = new byte[len];
                        if (!await ReadExactAsync(stream, payload, len, ct).ConfigureAwait(false)) break;
 
                        IpcEnvelope request;
                        try
                        {
                            request = BrokerIpc.DecodeEnvelope(Encoding.UTF8.GetString(payload));
                        }
                        catch (Exception)
                        {
                            break; // 协议失步，断开该连接
                        }
 
                        IpcEnvelope response = await DispatchAsync(request, ct).ConfigureAwait(false);
                        var frame = BrokerIpc.EncodeFrame(BrokerIpc.EncodeEnvelope(response));
                        await stream.WriteAsync(frame, 0, frame.Length, ct).ConfigureAwait(false);
                        await stream.FlushAsync(ct).ConfigureAwait(false);
                    }
                }
            }
            catch (OperationCanceledException) { }
            catch (IOException) { }
            catch (Exception ex)
            {
                Log.Warn($"[Broker] 连接处理结束: {ex.Message}");
            }
        }
 
        private async Task<IpcEnvelope> DispatchAsync(IpcEnvelope request, CancellationToken ct)
        {
            if (request.ProtocolVersion != BrokerIpc.ProtocolVersion)
                return Reply(request, BrokerIpc.SerializeBody(new BrokerIpc.ResultBody
                {
                    State = "Failed",
                    ErrorCode = "PROTOCOL_MISMATCH",
                    Readback = $"Broker 协议 v{BrokerIpc.ProtocolVersion}，拒绝 v{request.ProtocolVersion}"
                }));
 
            switch (request.MessageType)
            {
                case IpcMessageTypes.Hello:
                    return Reply(request, BrokerIpc.SerializeBody(new HelloResult
                    {
                        ServerEpoch = _lease.CurrentEpoch,
                        ConfigHash = CurrentHash(),
                        ProtocolVersion = BrokerIpc.ProtocolVersion,
                        BrokerId = _brokerId
                    }));
 
                case IpcMessageTypes.SubmitTask:
                    var body = BrokerIpc.DeserializeBody<BrokerIpc.SubmitBody>(request.BodyJson);
                    if (body == null)
                        return Reply(request, Result("Failed", "BAD_BODY", "SubmitBody 解析失败"));
                    return Reply(request, await ExecuteOneShotAsync(body, request, ct).ConfigureAwait(false));
 
                case IpcMessageTypes.ProbeReadback:
                    var pb = BrokerIpc.DeserializeBody<BrokerIpc.SubmitBody>(request.BodyJson);
                    return Reply(request, await ProbeReadbackAsync(pb?.DeviceId ?? "").ConfigureAwait(false));
 
                case IpcMessageTypes.CancelTask:
                    // MVP 不支持中断在途原子片段（INV-06：不得假装已取消）
                    return Reply(request, Result("Failed", "NOT_SUPPORTED", "MVP 不支持取消；片段将在 deadline 内完成"));
 
                default:
                    return Reply(request, Result("Failed", "UNKNOWN_MESSAGE", request.MessageType));
            }
        }
 
        // ==================== OneShot 执行核心 ====================
 
        private async Task<BrokerIpc.ResultBody> ExecuteOneShotAsync(
            BrokerIpc.SubmitBody body, IpcEnvelope request, CancellationToken ct)
        {
            // 0) OpCode 白名单
            if (!AllowedOps.Contains(body.OperationCode) ||
                !_devices.TryGetValue(body.DeviceId, out var device))
            {
                return Result("Failed", "DEVICE_NOT_BROKERED",
                    $"设备 {body.DeviceId} 不属于本 Broker 域或操作 {body.OperationCode} 不在白名单");
            }
 
            // 1) INV-04：configHash 对账（空哈希视为旧客户端，放行并记录）
            var hash = CurrentHash();
            if (!string.IsNullOrWhiteSpace(request.ConfigHash) &&
                !string.Equals(request.ConfigHash, hash, StringComparison.OrdinalIgnoreCase))
            {
                return Result("RejectedConfigMismatch", "CONFIG_MISMATCH",
                    $"客户端快照 {Trim(request.ConfigHash)} ≠ Broker 生效 {Trim(hash)}，先 Apply 再 I/O");
            }
 
            // 2) 栅栏再校验：CONNECT/DISCONNECT 仅登记或释放宿主物理会话、无总线输出，豁免租约；
            //    纯读回收邮箱（RECV/RECV_ALL/LOG/BUS_STATUS）为被动监控无写副作用，也豁免（支持调试页只读监听）；
            //    SEND 及其它 QUERY（可能触发设备行为，如 AUTO_RESPONSE/FILTER）必须有活跃租约（INV-01）
            string leaseKey = string.IsNullOrWhiteSpace(body.ResourceKey) ? body.DeviceId : body.ResourceKey;
            bool readOnlyDrain = body.OperationCode == "QUERY" &&
                (body.Payload.StartsWith("RECV", StringComparison.OrdinalIgnoreCase) ||
                 body.Payload.StartsWith("LOG", StringComparison.OrdinalIgnoreCase) ||
                 body.Payload.StartsWith("BUS_STATUS", StringComparison.OrdinalIgnoreCase) ||
                 body.Payload.StartsWith("STATUS", StringComparison.OrdinalIgnoreCase));
            if (body.OperationCode != "DISCONNECT" && body.OperationCode != "CONNECT" && !readOnlyDrain)
            {
                var phase = _lease.PhaseOf(leaseKey);
                if (phase == LeasePhase.RecoveryRequired)
                {
                    return Result("RejectedStaleLease", "RECOVERY_REQUIRED",
                        $"资源 {leaseKey} 过期待回收(INV-03)，禁止 I/O");
                }
                var holder = _lease.ProbeConflict(leaseKey);
                if (phase != LeasePhase.Active || holder == null)
                {
                    return Result("RejectedStaleLease", "NO_ACTIVE_LEASE",
                        $"资源 {leaseKey} 无活跃租约，客户端须先 Acquire（禁止无租约直接 I/O）");
                }
                // MVP：同机 arb.db 弱身份校验——只做存在性/相位检查；
                // owner↔clientInstanceId 精确对齐待 epoch 签名凭据（V3.0 §13 P2）。
            }
 
            // 3) SessionGate：同一物理会话原子的“寻道+执行”片段串行
            var gate = _gates.GetOrAdd(device.SessionKey, _ => new SemaphoreSlim(1, 1));
            bool entered;
            try { entered = await gate.WaitAsync(5000, CancellationToken.None).ConfigureAwait(false); }
            catch (ObjectDisposedException) { entered = false; }
            if (!entered)
            {
                // 未进入片段 → 确定未执行，可安全判忙
                return Result("Failed", "RESOURCE_BUSY",
                    $"会话 {device.SessionKey} 片段队列等待超时，未执行可安全重试");
            }
 
            try
            {
                return await RunFragmentAsync(device, body, ct).ConfigureAwait(false);
            }
            finally
            {
                gate.Release();
            }
        }
 
        private async Task<BrokerIpc.ResultBody> RunFragmentAsync(
            ManagedDevice device, BrokerIpc.SubmitBody body, CancellationToken ct)
        {
            try
            {
                if (body.OperationCode == "CONNECT" || (device.Driver == null && body.OperationCode != "DISCONNECT"))
                {
                    await EnsureConnectedAsync(device, ct).ConfigureAwait(false);
                }
 
                switch (body.OperationCode)
                {
                    case "CONNECT":
                        return Result("Succeeded", "", $"OK:CONNECTED:{device.DeviceId}", _lease.CurrentEpoch);
                    case "DISCONNECT":
                        if (device.Connected)
                        {
                            await device.Driver!.DisconnectAsync().ConfigureAwait(false);
                            device.Connected = false;
                        }
                        return Result("Succeeded", "", $"OK:DISCONNECTED:{device.DeviceId}", _lease.CurrentEpoch);
                    case "SEND":
                        {
                            EnsureConnectedOrThrow(device);
                            var resp = await device.Driver!.SendCommandAsync(body.Payload, ct).ConfigureAwait(false);
                            return Result("Succeeded", "", AppendPushes(device, resp), _lease.CurrentEpoch);
                        }
                    case "QUERY":
                        {
                            EnsureConnectedOrThrow(device);
                            var resp = await device.Driver!.QueryAsync(body.Payload, ct).ConfigureAwait(false);
                            return Result("Succeeded", "", AppendPushes(device, resp), _lease.CurrentEpoch);
                        }
                    case "STATUS":
                        {
                            var st = device.Driver?.GetStatus();
                            return Result("Succeeded", "",
                                st == null ? "STATUS:NOT_CREATED"
                                    : $"STATUS:{st.IsConnected}|{st.LastError}", _lease.CurrentEpoch);
                        }
                    default:
                        return Result("Failed", "BAD_OP", body.OperationCode);
                }
            }
            catch (OperationCanceledException)
            {
                // 片段中途取消：物理侧可能已执行 → INV-06
                return Result("Unknown", "DEADLINE", $"{body.OperationCode} 超 deadline，结果未知");
            }
            catch (Exception ex)
            {
                bool ioMayHappened = body.OperationCode is "SEND" or "QUERY";
                return Result(ioMayHappened ? "Unknown" : "Failed", "DRIVER_ERROR",
                    $"{body.OperationCode} 执行异常: {ex.Message}", _lease.CurrentEpoch);
            }
        }
 
        /// <summary>
        /// 捎带推送（§5.4）：将宿主物理驱动自上次响应以来收到的 EV:RECV 帧，
        /// 以 "|PUSHED:n|EV:RECV…" 后缀拼到响应 readback，客户端 BrokerRpcTransport 解析上屏。
        /// </summary>
        private static string AppendPushes(ManagedDevice device, string resp)
        {
            if (device.Driver is CanDeviceDriver can)
            {
                var lines = can.DrainPendingPushLines();
                if (lines.Count > 0)
                    return resp + "|PUSHED:" + lines.Count + "|" + string.Join("|", lines);
            }
            return resp;
        }
 
        private void EnsureConnectedOrThrow(ManagedDevice device)
        {
            if (!device.Connected || device.Driver == null)
                throw new InvalidOperationException($"设备 {device.DeviceId} 未连接，请先 CONNECT");
        }
 
        private async Task EnsureConnectedAsync(ManagedDevice device, CancellationToken ct)
        {
            if (device.Connected && device.Driver != null) return;
            await _driverCreateLock.WaitAsync(ct).ConfigureAwait(false);
            try
            {
                if (device.Connected && device.Driver != null) return;
                device.Driver ??= DriverFactory.CreateDriver(device.DeviceId, device.Transport, device.Endpoint);
                await device.Driver.ConnectAsync(ct).ConfigureAwait(false);
                device.Connected = true;
                Log.Info($"[Broker] 物理会话已建立: {device.DeviceId} ({device.Transport} {device.Endpoint})");
            }
            finally
            {
                _driverCreateLock.Release();
            }
        }
 
        /// <summary>恢复回收前读回探测：宿主读回设备状态作为合格证据（INV-03 人工回收链的 Broker 侧）。</summary>
        private async Task<BrokerIpc.ResultBody> ProbeReadbackAsync(string deviceId)
        {
            if (!_devices.TryGetValue(deviceId, out var device) || device.Driver == null)
                return Result("Failed", "NO_DRIVER", $"设备 {deviceId} 无活动驱动");
            try
            {
                var readback = await device.Driver.QueryAsync("STATUS", CancellationToken.None).ConfigureAwait(false);
                return Result("Succeeded", "", readback, _lease.CurrentEpoch);
            }
            catch (Exception ex)
            {
                return Result("Failed", "READBACK_ERROR", ex.Message);
            }
        }
 
        // ==================== 杂项 ====================
 
        private string CurrentHash()
        {
            long now = BrokerIpc.NowUtcMs();
            if (now - Interlocked.Read(ref _hashStampMs) > 2000)
            {
                try
                {
                    _cachedHash = DeviceConfigHash.Compute(_configDbPath);
                }
                catch (Exception ex)
                {
                    Log.Warn($"[Broker] configHash 重算失败: {ex.Message}");
                }
                Interlocked.Exchange(ref _hashStampMs, now);
            }
            return _cachedHash;
        }
 
        private static BrokerIpc.ResultBody Result(string state, string code, string readback, long epoch = 0)
            => new BrokerIpc.ResultBody
            {
                State = state,
                ErrorCode = code,
                Readback = readback,
                ServerEpoch = epoch
            };
 
        private static IpcEnvelope Reply(IpcEnvelope request, string bodyJson) => new IpcEnvelope
        {
            ProtocolVersion = BrokerIpc.ProtocolVersion,
            MessageType = "Result",
            RequestId = Guid.NewGuid().ToString("N"),
            CorrelationId = request.RequestId,
            ClientInstanceId = request.ClientInstanceId,
            AuthorityId = request.AuthorityId,
            BodyJson = bodyJson
        };
 
        private static IpcEnvelope Reply(IpcEnvelope request, BrokerIpc.ResultBody body)
            => Reply(request, BrokerIpc.SerializeBody(body));
 
        private static async Task<bool> ReadExactAsync(Stream s, byte[] buf, int count, CancellationToken ct)
        {
            int off = 0;
            while (off < count)
            {
                int n;
                try { n = await s.ReadAsync(buf, off, count - off, ct).ConfigureAwait(false); }
                catch (IOException) { return false; }
                if (n <= 0) return false;
                off += n;
            }
            return true;
        }
 
        private static string Trim(string h)
            => string.IsNullOrEmpty(h) ? "" : (h.Length > 8 ? h.Substring(0, 8) : h);
    }
}