using System;
using System.Buffers.Binary;
using System.Collections.Concurrent;
using System.IO;
using System.IO.Pipes;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Aging.ResourceContracts;
using log4net;

namespace Aging.Infrastructure.Broker
{
    /// <summary>
    /// DeviceBroker 命名管道客户端通道（V3.0 §13 同机基线）。
    /// 修复现有 PipeClient 的三大问题：响应关联可达、超时生效、取消可传播。
    /// 帧格式：4 字节大端长度前缀 + UTF8 JSON 封套；请求-响应经 RequestId/CorrelationId 配对。
    /// 线程模型：单后台读循环 + pending 字典；写入用信号量串行化。线程安全，可被多租约共享。
    /// </summary>
    public sealed class BrokerClientChannel : IDisposable
    {
        private static readonly ILog Log = LogManager.GetLogger(typeof(BrokerClientChannel));

        private readonly string _pipeName;
        private readonly string _clientInstanceId;
        private readonly SemaphoreSlim _writeLock = new SemaphoreSlim(1, 1);
        private readonly ConcurrentDictionary<string, TaskCompletionSource<IpcEnvelope>> _pending = new();

        private NamedPipeClientStream? _pipe;
        private Task? _readLoop;
        private volatile bool _disposed;
        private volatile bool _connected;
        private int _inFlight;

        /// <summary>握手成功后的 Broker 纪元（重启后递增，用于本地陈旧判定）</summary>
        public long ServerEpoch { get; private set; }

        /// <summary>Broker 生效配置哈希（INV-04 对账参照）</summary>
        public string BrokerConfigHash { get; private set; } = "";

        public string BrokerId { get; private set; } = "";

        public bool IsConnected => _connected && !_disposed;

        /// <summary>连接/握手默认超时（毫秒）</summary>
        public int ConnectTimeoutMs { get; set; } = 3000;

        /// <summary>单次 RPC 默认等待上限（毫秒）</summary>
        public int DefaultRequestTimeoutMs { get; set; } = 10_000;

        public BrokerClientChannel(string pipeName, string clientInstanceId)
        {
            _pipeName = pipeName ?? throw new ArgumentNullException(nameof(pipeName));
            _clientInstanceId = clientInstanceId ?? throw new ArgumentNullException(nameof(clientInstanceId));
        }

        /// <summary>
        /// 确保连接并完成 Hello 握手（幂等；断线后可重呼）。
        /// 协议主版本不匹配时抛出 InvalidOperationException 并拒绝后续请求。
        /// </summary>
        public async Task EnsureConnectedAsync(CancellationToken ct)
        {
            if (_disposed) throw new ObjectDisposedException(nameof(BrokerClientChannel));
            if (_connected) return;

            var pipe = new NamedPipeClientStream("localhost", _pipeName,
                PipeDirection.InOut, PipeOptions.Asynchronous);
            try
            {
                using (var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(ct))
                {
                    timeoutCts.CancelAfter(ConnectTimeoutMs);
                    await pipe.ConnectAsync(timeoutCts.Token).ConfigureAwait(false);
                }
            }
            catch (Exception ex)
            {
                pipe.Dispose();
                throw new InvalidOperationException($"BROKER_UNREACHABLE: 无法连接 DeviceBroker 管道 [{_pipeName}]: {ex.Message}", ex);
            }

            _pipe = pipe;
            _connected = true;
            _readLoop = Task.Run(() => ReadLoopAsync(pipe, ct));

            // Hello 握手
            var hello = NewEnvelope(IpcMessageTypes.Hello);
            var resp = await SendAsync(hello, ct).ConfigureAwait(false);
            var result = BrokerIpc.DeserializeBody<HelloResult>(resp.BodyJson);
            if (result == null)
                throw new InvalidOperationException("BROKER_HANDSHAKE_FAILED: Hello 响应无效");
            if (result.ProtocolVersion != BrokerIpc.ProtocolVersion)
            {
                _connected = false;
                throw new InvalidOperationException(
                    $"BROKER_PROTOCOL_MISMATCH: 客户端 v{BrokerIpc.ProtocolVersion}, Broker v{result.ProtocolVersion}");
            }
            ServerEpoch = result.ServerEpoch;
            BrokerConfigHash = result.ConfigHash;
            BrokerId = result.BrokerId;
            Log.Info($"[BrokerChannel] 已连接 {_pipeName}: BrokerId={BrokerId}, epoch={ServerEpoch}, configHash={ShortHash(BrokerConfigHash)}");
        }

        /// <summary>
        /// 发送一帧请求并等待其关联响应。超时/断线均返回可区分的错误封套（不抛裸异常）。
        /// </summary>
        public async Task<IpcEnvelope> SendAsync(IpcEnvelope request, CancellationToken ct)
        {
            if (_disposed) throw new ObjectDisposedException(nameof(BrokerClientChannel));
            var pipe = _pipe;
            if (!_connected || pipe == null)
                return ErrorEnvelope(request, "BROKER_UNREACHABLE", "未连接 DeviceBroker");

            var tcs = new TaskCompletionSource<IpcEnvelope>(TaskCreationOptions.RunContinuationsAsynchronously);
            _pending[request.RequestId] = tcs;
            Interlocked.Increment(ref _inFlight);
            try
            {
                byte[] frame;
                try
                {
                    frame = BrokerIpc.EncodeFrame(BrokerIpc.EncodeEnvelope(request));
                }
                catch (Exception ex)
                {
                    return ErrorEnvelope(request, "PROTOCOL_ERROR", ex.Message);
                }

                // 写入串行化；pipe 半双工交错下 async 写可能挂死，故写入不加应用级超时（依赖读侧 deadline + 断线感知）
                await _writeLock.WaitAsync(ct).ConfigureAwait(false);
                try
                {
                    await pipe.WriteAsync(frame, 0, frame.Length, ct).ConfigureAwait(false);
                    await pipe.FlushAsync(ct).ConfigureAwait(false);
                }
                catch (IOException ex)
                {
                    Fault($"管道写失败: {ex.Message}");
                    return ErrorEnvelope(request, "BROKER_IO_ERROR", ex.Message);
                }
                finally
                {
                    _writeLock.Release();
                }

                using (var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(ct))
                {
                    int waitMs = request.DeadlineUtcMs > 0
                        ? (int)Math.Max(1, request.DeadlineUtcMs - BrokerIpc.NowUtcMs())
                        : DefaultRequestTimeoutMs;
                    var delay = Task.Delay(waitMs, timeoutCts.Token);
                    var winner = await Task.WhenAny(tcs.Task, delay).ConfigureAwait(false);
                    if (winner == tcs.Task)
                    {
                        timeoutCts.Cancel();  // 释放 delay 定时器
                        return await tcs.Task.ConfigureAwait(false);
                    }
                    // INV-06：请求已写出且可能在途 → 结果未知(Unknown)，禁止当未执行
                    return UnknownEnvelope(request, "BROKER_TIMEOUT",
                        "等待响应超时，结果未知，按未完成处理");
                }
            }
            finally
            {
                _pending.TryRemove(request.RequestId, out _);
                Interlocked.Decrement(ref _inFlight);
            }
        }

        /// <summary>便捷方法：提交一次 OneShot 设备动作，返回结果体。</summary>
        public async Task<BrokerIpc.ResultBody> SubmitAsync(BrokerIpc.SubmitBody body, string configHash, CancellationToken ct)
        {
            await EnsureConnectedAsync(ct).ConfigureAwait(false);
            var env = NewEnvelope(IpcMessageTypes.SubmitTask);
            env.ConfigHash = configHash ?? "";
            env.BodyJson = BrokerIpc.SerializeBody(body);
            var resp = await SendAsync(env, ct).ConfigureAwait(false);
            var result = BrokerIpc.DeserializeBody<BrokerIpc.ResultBody>(resp.BodyJson)
                ?? new BrokerIpc.ResultBody { State = "Unknown", ErrorCode = "BAD_RESPONSE" };
            return result;
        }

        // ==================== 内部 ====================

        private async Task ReadLoopAsync(NamedPipeClientStream pipe, CancellationToken ct)
        {
            var lenBuf = new byte[4];
            try
            {
                while (!ct.IsCancellationRequested && !_disposed)
                {
                    await ReadExactAsync(pipe, lenBuf, 4, ct).ConfigureAwait(false);
                    int len = BinaryPrimitives.ReadInt32BigEndian(lenBuf);
                    if (len <= 0 || len > BrokerIpc.MaxFrameBytes)
                        throw new IOException($"非法帧长度 {len}");
                    var payload = new byte[len];
                    await ReadExactAsync(pipe, payload, len, ct).ConfigureAwait(false);

                    IpcEnvelope env;
                    try
                    {
                        env = BrokerIpc.DecodeEnvelope(Encoding.UTF8.GetString(payload));
                    }
                    catch (Exception)
                    {
                        continue; // 无法解析的单帧丢弃，不影响后续关联
                    }

                    if (!string.IsNullOrEmpty(env.CorrelationId)
                        && _pending.TryGetValue(env.CorrelationId, out var tcs))
                    {
                        tcs.TrySetResult(env);
                    }
                }
            }
            catch (OperationCanceledException) { /* 正常退出 */ }
            catch (Exception ex)
            {
                if (!_disposed) Fault($"读循环终止: {ex.Message}");
            }
        }

        private static async Task ReadExactAsync(Stream s, byte[] buf, int count, CancellationToken ct)
        {
            int off = 0;
            while (off < count)
            {
                int n = await s.ReadAsync(buf, off, count - off, ct).ConfigureAwait(false);
                if (n <= 0) throw new EndOfStreamException("管道已关闭");
                off += n;
            }
        }

        /// <summary>连接失效：置降级态，等待下次 EnsureConnectedAsync 重连；在途请求由其自身超时收口。</summary>
        private void Fault(string reason)
        {
            if (!_connected) return;
            _connected = false;
            Log.Warn($"[BrokerChannel] 连接失效({reason})，将在下次请求时重连");
            try { _pipe?.Dispose(); } catch { }
            _pipe = null;
        }

        private IpcEnvelope NewEnvelope(string type) => new IpcEnvelope
        {
            ProtocolVersion = BrokerIpc.ProtocolVersion,
            MessageType = type,
            RequestId = Guid.NewGuid().ToString("N"),
            CorrelationId = "",
            ClientInstanceId = _clientInstanceId,
            AuthorityId = BrokerId
        };

        private static IpcEnvelope ErrorEnvelope(IpcEnvelope? request, string code, string message)
            => BuildEnvelope(request, "Failed", code, message);

        private static IpcEnvelope UnknownEnvelope(IpcEnvelope? request, string code, string message)
            => BuildEnvelope(request, "Unknown", code, message);

        private static IpcEnvelope BuildEnvelope(IpcEnvelope? request, string state, string code, string message)
            => new IpcEnvelope
            {
                ProtocolVersion = BrokerIpc.ProtocolVersion,
                MessageType = "Result",
                CorrelationId = request?.RequestId ?? "",
                BodyJson = "{\"State\":\"" + state + "\",\"ErrorCode\":\"" + code + "\",\"Readback\":\"" +
                           message.Replace("\"", "'") + "\"}"
            };

        private static string ShortHash(string h)
            => string.IsNullOrEmpty(h) ? "" : (h.Length > 8 ? h.Substring(0, 8) : h);

        public void Dispose()
        {
            if (_disposed) return;
            _disposed = true;
            _connected = false;
            try { _pipe?.Dispose(); } catch { }
            _pipe = null;
            _writeLock.Dispose();
        }
    }
}