using System.Buffers;
using System.Collections.Concurrent;
using System.Text;
using NewLife.Data;
using NewLife.Log;
using NewLife.Serialization;

namespace NewLife.Messaging;

/// <summary>事件分发枢纽。按主题路由网络消息或本地事件到各事件总线</summary>
/// <remarks>
/// <para><b>核心职责</b>：</para>
/// <list type="number">
/// <item><description><b>主题路由</b>：维护 <c>topic → IEventBus&lt;TEvent&gt;</c> 映射，提供 <see cref="GetEventBus"/>/<see cref="RegisterBus"/>。</description></item>
/// <item><description><b>协议适配</b>：默认解析 <c>event#topic#clientId#message</c> 文本协议；
/// 派生类可重写 <see cref="TryDecode(IPacket,out EventEnvelope)"/> / <see cref="TryDecode(String,out EventEnvelope)"/> 替换为二进制、MQTT 等协议。</description></item>
/// <item><description><b>控制面 + 数据面</b>：解析后区分动作（订阅/取消订阅）与数据消息，分别走 <see cref="SubscribeAsync"/>/<see cref="UnsubscribeAsync"/> 或 <see cref="PublishAsync"/>。</description></item>
/// </list>
/// <para><b>典型应用</b>：作为 IoT 网关或 MQ Bridge，将网络层（如 WebSocket/TCP）收到的消息按 topic 投递到进程内订阅者。</para>
/// </remarks>
/// <typeparam name="TEvent">事件类型</typeparam>
public class EventHub<TEvent> : IEventHandler<IPacket>, IEventHandler<String>, ILogFeature, ITracerFeature
{
    #region 嵌套：事件信封
    /// <summary>协议解码的中间结果，承载主题/客户端/动作或事件三种语义</summary>
    protected readonly struct EventEnvelope
    {
        /// <summary>主题</summary>
        public String Topic { get; }

        /// <summary>客户端标识。发送方，路由分发时用于排除回环</summary>
        public String ClientId { get; }

        /// <summary>动作。订阅/取消订阅控制指令；为空表示数据消息</summary>
        public String? Action { get; }

        /// <summary>已解码的事件实例</summary>
        public TEvent? Event { get; }

        /// <summary>是否动作信封</summary>
        public Boolean IsAction => !Action.IsNullOrEmpty();

        /// <summary>构造动作信封</summary>
        /// <param name="topic">主题</param>
        /// <param name="clientId">客户端标识</param>
        /// <param name="action">动作指令</param>
        public static EventEnvelope ForAction(String topic, String clientId, String action) => new(topic, clientId, action, default);

        /// <summary>构造事件信封</summary>
        /// <param name="topic">主题</param>
        /// <param name="clientId">客户端标识</param>
        /// <param name="event">事件实例</param>
        public static EventEnvelope ForEvent(String topic, String clientId, TEvent @event) => new(topic, clientId, null, @event);

        private EventEnvelope(String topic, String clientId, String? action, TEvent? @event)
        {
            Topic = topic;
            ClientId = clientId;
            Action = action;
            Event = @event;
        }
    }
    #endregion

    #region 属性
    /// <summary>事件总线工厂。用于按需创建各 topic 对应的事件总线</summary>
    public IEventBusFactory? Factory { get; set; }

    /// <summary>JSON 主机。用于编解码事件体</summary>
    public IJsonHost JsonHost { get; set; } = JsonHelper.Default;

    /// <summary>JSON序列化选项，影响复杂对象的编码和解码行为</summary>
    public JsonOptions? JsonOptions { get; set; }

    /// <summary>动作判定阈值。消息体短于该长度且不以 <c>{</c> 开头时视为控制动作指令</summary>
    public Int32 ActionMaxLength { get; set; } = 32;

    /// <summary>链路追踪</summary>
    public ITracer? Tracer { get; set; }

    private readonly ConcurrentDictionary<String, IEventBus<TEvent>> _eventBuses = new();
    /// <summary>已注册的事件总线集合。Key 为 topic</summary>
    public IDictionary<String, IEventBus<TEvent>> EventBuses => _eventBuses;

    private static readonly Byte[] _prefixBytes = Encoding.ASCII.GetBytes("event#");
    private static readonly Char[] _prefixChars = "event#".ToCharArray();
    #endregion

    #region 注册与获取
    /// <summary>注册主题对应的事件总线</summary>
    /// <param name="topic">事件主题</param>
    /// <param name="eventBus">事件总线实例</param>
    public void RegisterBus(String topic, IEventBus<TEvent> eventBus) => _eventBuses[topic] = eventBus;

    /// <summary>尝试获取主题对应的事件总线（不会创建新的）</summary>
    /// <param name="topic">事件主题</param>
    /// <param name="eventBus">事件总线</param>
    /// <returns>是否存在</returns>
    public Boolean TryGetBus(String topic, out IEventBus<TEvent> eventBus) => _eventBuses.TryGetValue(topic, out eventBus!);

    /// <summary>获取或创建主题对应的事件总线</summary>
    /// <param name="topic">事件主题</param>
    /// <param name="clientId">客户标识，仅在使用 <see cref="Factory"/> 创建时传递</param>
    /// <returns>事件总线实例</returns>
    public virtual IEventBus<TEvent> GetEventBus(String topic, String clientId = "")
    {
        if (_eventBuses.TryGetValue(topic, out var bus)) return bus;
        bus = Factory?.CreateEventBus<TEvent>(topic, clientId) ?? new EventBus<TEvent>();
        return _eventBuses.GetOrAdd(topic, bus);
    }
    #endregion

    #region 发布与订阅
    /// <summary>向指定主题发布事件</summary>
    /// <remarks>
    /// 发送方信息（用于排除回环）通过 <see cref="EventContext.ClientId"/> 传递。
    /// 无需排除回环时可传入 <see langword="null"/>，由总线自动创建上下文。
    /// </remarks>
    /// <param name="topic">主题</param>
    /// <param name="event">事件</param>
    /// <param name="context">事件上下文；为空时由总线自行创建</param>
    /// <param name="cancellationToken">取消令牌</param>
    /// <returns>成功处理该事件的处理器数量</returns>
    public virtual Task<Int32> PublishAsync(String topic, TEvent @event, IEventContext? context = null, CancellationToken cancellationToken = default)
    {
        var bus = GetEventBus(topic);

        // 自动注入 Topic 到上下文，便于订阅者读取
        if (context is EventContext ctx) ctx.Topic ??= topic;

        return bus.PublishAsync(@event, context, cancellationToken);
    }

    /// <summary>向指定主题订阅事件</summary>
    /// <param name="topic">主题</param>
    /// <param name="clientId">客户标识</param>
    /// <param name="handler">事件处理器</param>
    /// <param name="cancellationToken">取消令牌</param>
    /// <returns>是否订阅成功</returns>
    public virtual Task<Boolean> SubscribeAsync(String topic, String clientId, IEventHandler<TEvent> handler, CancellationToken cancellationToken = default) => GetEventBus(topic, clientId).SubscribeAsync(handler, clientId, cancellationToken);

    /// <summary>取消指定主题/客户端的订阅</summary>
    /// <param name="topic">主题</param>
    /// <param name="clientId">客户标识</param>
    /// <param name="cancellationToken">取消令牌</param>
    /// <returns>是否成功取消订阅</returns>
    public virtual async Task<Boolean> UnsubscribeAsync(String topic, String clientId, CancellationToken cancellationToken = default)
    {
        if (!_eventBuses.TryGetValue(topic, out var bus)) return false;

        var ok = await bus.UnsubscribeAsync(clientId, cancellationToken).ConfigureAwait(false);

        // 取消后若已无订阅者且为默认实现，移除总线避免泄漏。
        // 采用"先移除再复查放回"缩小检查-再行动窗口：若移除后恰好有并发订阅注册到该总线，则重新放回
        if (bus is EventBus<TEvent> eb && eb.Handlers.Count == 0)
        {
            if (_eventBuses.TryRemove(topic, out _) && eb.Handlers.Count > 0)
                _eventBuses.TryAdd(topic, eb);
        }

        return ok;
    }
    #endregion

    #region 协议解码（可重载）
    /// <summary>尝试从二进制数据包解码事件信封。派生类可重写以替换协议实现</summary>
    /// <param name="data">网络数据包</param>
    /// <param name="envelope">解码出的事件信封</param>
    /// <returns>是否解码成功</returns>
    protected virtual Boolean TryDecode(IPacket data, out EventEnvelope envelope)
    {
        envelope = default;
        if (data == null) return false;

        if (data.Next == null)
        {
            if (!TryParseHeader(data.GetSpan(), out var topic, out var clientId, out var headerLen)) return false;

            return DecodePayload(data, topic, clientId, headerLen, out envelope);
        }
        else
        {
            // 头部 event#topic#clientId# 可能跨节点（跨接收轮组链的帧）：跨段扫描定位头部末端，物化头部字节后复用跨度解析
            var headLength = FindHeaderLength(data);
            if (headLength <= 0) return false;

            if (!TryParseHeader(data.ReadBytes(0, headLength), out var topic, out var clientId, out var headerLen)) return false;

            return DecodePayload(data, topic, clientId, headerLen, out envelope);
        }
    }

    /// <summary>跨段扫描头部末端。event#topic#clientId# 的分隔符为单字节 '#'，用序列读取器跨段扫描</summary>
    /// <param name="data">数据包链</param>
    /// <returns>头部字节数（含末尾 '#'）；分隔符不足 3 个时返回 0</returns>
    private static Int32 FindHeaderLength(IPacket data)
    {
        var reader = new SequenceReader<Byte>(data.AsReadOnlySequence());
        for (var i = 0; i < 3; i++)
        {
            if (!reader.TryAdvanceTo((Byte)'#', true)) return 0;
        }

        return (Int32)reader.Consumed;
    }

    /// <summary>从头部之后的负载解码事件信封</summary>
    /// <param name="data">网络数据包</param>
    /// <param name="topic">主题</param>
    /// <param name="clientId">客户端标识</param>
    /// <param name="headerLen">头部字节数</param>
    /// <param name="envelope">解码出的事件信封</param>
    /// <returns>是否解码成功</returns>
    /// <remarks>
    /// <para>当 <typeparamref name="TEvent"/> 本身是数据包时，负载切片所有权随之转移给订阅者（订阅者负责释放）；
    /// 其它事件类型则切片仅供取值，本方法内部归还引用计数。</para>
    /// </remarks>
    private Boolean DecodePayload(IPacket data, String topic, String clientId, Int32 headerLen, out EventEnvelope envelope)
    {
        envelope = default;

        var msg = data.Slice(headerLen);
        if (msg.Length == 0)
        {
            msg.TryDispose();

            return false;
        }

        // TEvent 本身是 IPacket，直接零拷贝构造事件信封（切片随信封交给订阅者释放）
        if (msg is TEvent evt) { envelope = EventEnvelope.ForEvent(topic, clientId, evt); return true; }

        // 其它事件类型：切片只用于取值，取值后立即归还引用计数（不归还则缓冲永远回不了池）
        try
        {
            var msgStr = msg.ToStr();

            return BuildEnvelopeFromString(topic, clientId, msgStr, out envelope);
        }
        finally
        {
            msg.TryDispose();
        }
    }

    /// <summary>尝试从字符串解码事件信封。派生类可重写以替换协议实现</summary>
    /// <param name="data">字符串消息</param>
    /// <param name="envelope">解码出的事件信封</param>
    /// <returns>是否解码成功</returns>
    protected virtual Boolean TryDecode(String data, out EventEnvelope envelope)
    {
        envelope = default;
        if (data.IsNullOrEmpty()) return false;
        if (!TryParseHeader(data.AsSpan(), out var topic, out var clientId, out var headerLen)) return false;

        var msg = data[headerLen..];
        if (msg.Length == 0) return false;

        return BuildEnvelopeFromString(topic, clientId, msg, out envelope);
    }

    /// <summary>编码事件为字符串（便于通过文本协议发送）</summary>
    /// <param name="topic">主题</param>
    /// <param name="clientId">客户端标识</param>
    /// <param name="event">事件实例</param>
    /// <returns>编码后的字符串</returns>
    protected virtual String EncodeEvent(String topic, String clientId, TEvent @event)
    {
        var body = @event is String s ? s : JsonHost.Write(@event!, JsonOptions);
        return $"event#{topic}#{clientId}#{body}";
    }

    /// <summary>编码控制动作为字符串（订阅/取消订阅等）</summary>
    /// <param name="topic">主题</param>
    /// <param name="clientId">客户端标识</param>
    /// <param name="action">动作指令</param>
    /// <returns>编码后的字符串</returns>
    protected virtual String EncodeAction(String topic, String clientId, String action) => $"event#{topic}#{clientId}#{action}";

    /// <summary>根据消息体字符串构造事件信封：动作 / 字符串事件 / JSON 解码</summary>
    private Boolean BuildEnvelopeFromString(String topic, String clientId, String msg, out EventEnvelope envelope)
    {
        // TEvent = String，整条消息体始终视为事件，不识别动作（避免短消息被误判为控制指令）
        if (msg is TEvent strEvt)
        {
            envelope = EventEnvelope.ForEvent(topic, clientId, strEvt);
            return true;
        }

        // 短字符串且不以 { 开头 → 视为动作指令
        if (msg[0] != '{' && msg.Length < ActionMaxLength)
        {
            envelope = EventEnvelope.ForAction(topic, clientId, msg);
            return true;
        }

        // JSON 反序列化
        var evt = JsonHost.Read<TEvent>(msg, JsonOptions);
        if (evt == null)
        {
            envelope = default;
            return false;
        }
        envelope = EventEnvelope.ForEvent(topic, clientId, evt);
        return true;
    }

    /// <summary>解析 <c>event#topic#clientId#</c> 二进制头部</summary>
    /// <param name="data">输入数据</param>
    /// <param name="topic">输出主题</param>
    /// <param name="clientId">输出客户端标识</param>
    /// <param name="headerLength">头部字节数（含末尾 #）</param>
    /// <returns>是否解析成功</returns>
    public static Boolean TryParseHeader(ReadOnlySpan<Byte> data, out String topic, out String clientId, out Int32 headerLength)
    {
        topic = clientId = String.Empty;
        headerLength = 0;
        if (!data.StartsWith(_prefixBytes)) return false;

        var p = data.IndexOf((Byte)'#');
        var rest = data[(p + 1)..];
        var p2 = rest.IndexOf((Byte)'#');
        if (p2 <= 0) return false;
        topic = rest[..p2].ToStr();

        var rest2 = rest[(p2 + 1)..];
        var p3 = rest2.IndexOf((Byte)'#');
        if (p3 <= 0) return false;
        clientId = rest2[..p3].ToStr();

        headerLength = p + 1 + p2 + 1 + p3 + 1;
        return true;
    }

    /// <summary>解析 <c>event#topic#clientId#</c> 字符串头部</summary>
    /// <param name="data">输入数据</param>
    /// <param name="topic">输出主题</param>
    /// <param name="clientId">输出客户端标识</param>
    /// <param name="headerLength">头部字符数（含末尾 #）</param>
    /// <returns>是否解析成功</returns>
    public static Boolean TryParseHeader(ReadOnlySpan<Char> data, out String topic, out String clientId, out Int32 headerLength)
    {
        topic = clientId = String.Empty;
        headerLength = 0;
        if (!data.StartsWith(_prefixChars)) return false;

        var p = data.IndexOf('#');
        var rest = data[(p + 1)..];
        var p2 = rest.IndexOf('#');
        if (p2 <= 0) return false;
        topic = rest[..p2].ToString();

        var rest2 = rest[(p2 + 1)..];
        var p3 = rest2.IndexOf('#');
        if (p3 <= 0) return false;
        clientId = rest2[..p3].ToString();

        headerLength = p + 1 + p2 + 1 + p3 + 1;
        return true;
    }
    #endregion

    #region 网络消息接收
    /// <summary>接收网络字节流消息并按协议解码后路由</summary>
    /// <param name="data">网络数据包</param>
    /// <param name="context">事件上下文</param>
    /// <param name="cancellationToken">取消令牌</param>
    /// <returns>分发到本地订阅者的处理器数量</returns>
    public virtual async Task<Int32> OnReceiveAsync(IPacket data, IEventContext? context = null, CancellationToken cancellationToken = default)
    {
        if (data == null) return 0;
        if (!TryDecode(data, out var envelope)) return 0;

        return await DispatchEnvelopeAsync(envelope, data, context, cancellationToken).ConfigureAwait(false);
    }

    /// <summary>接收字符串消息并按协议解码后路由</summary>
    /// <param name="data">字符串消息</param>
    /// <param name="context">事件上下文</param>
    /// <param name="cancellationToken">取消令牌</param>
    /// <returns>分发到本地订阅者的处理器数量</returns>
    public virtual async Task<Int32> OnReceiveAsync(String data, IEventContext? context = null, CancellationToken cancellationToken = default)
    {
        if (data.IsNullOrEmpty()) return 0;
        if (!TryDecode(data, out var envelope)) return 0;

        return await DispatchEnvelopeAsync(envelope, data, context, cancellationToken).ConfigureAwait(false);
    }

    Task IEventHandler<IPacket>.HandleAsync(IPacket @event, IEventContext? context, CancellationToken cancellationToken) => OnReceiveAsync(@event, context, cancellationToken);
    Task IEventHandler<String>.HandleAsync(String @event, IEventContext? context, CancellationToken cancellationToken) => OnReceiveAsync(@event, context, cancellationToken);

    /// <summary>把解码后的事件信封路由到控制面或数据面</summary>
    private async Task<Int32> DispatchEnvelopeAsync(EventEnvelope envelope, Object raw, IEventContext? context, CancellationToken cancellationToken)
    {
        // 把原始数据透传到扩展项，便于网络层做后续路由（如转发到其他客户端）
        if (context is IExtend ext) ext["Raw"] = raw;

        // 控制面：订阅/取消订阅
        if (envelope.IsAction)
        {
            var action = envelope.Action!;
            if (action.EqualIgnoreCase("subscribe"))
            {
                if ((context as IExtend)?["Handler"] is IEventHandler<TEvent> handler)
                {
                    await SubscribeAsync(envelope.Topic, envelope.ClientId, handler, cancellationToken).ConfigureAwait(false);
                    return 1;
                }
                return 0;
            }
            if (action.EqualIgnoreCase("unsubscribe"))
            {
                await UnsubscribeAsync(envelope.Topic, envelope.ClientId, cancellationToken).ConfigureAwait(false);
                return 1;
            }
            return 0;
        }

        // 数据面：发布事件
        if (envelope.Event is null) return 0;

        // 将发送方 ClientId 注入上下文，便于本地总线排除回环
        if (context is EventContext ec)
        {
            ec.ClientId ??= envelope.ClientId;
        }
        else if (context == null && !envelope.ClientId.IsNullOrEmpty())
        {
            context = new EventContext { Topic = envelope.Topic, ClientId = envelope.ClientId };
        }

        return await PublishAsync(envelope.Topic, envelope.Event, context, cancellationToken).ConfigureAwait(false);
    }
    #endregion

    #region 等待接收
    /// <summary>异步等待指定主题的第一条事件</summary>
    /// <param name="topic">主题</param>
    /// <param name="cancellationToken">取消令牌</param>
    /// <returns>事件实例</returns>
    public async Task<TEvent> ReceiveAsync(String topic, CancellationToken cancellationToken = default)
    {
        var bus = GetEventBus(topic);
        try
        {
            return await bus.ReceiveAsync(cancellationToken).ConfigureAwait(false);
        }
        finally
        {
            // 等待完成后若无其他订阅者，移除总线避免泄漏。
            // 采用"先移除再复查放回"缩小检查-再行动窗口，避免并发订阅丢失
            if (bus is EventBus<TEvent> eb && eb.Handlers.Count == 0)
            {
                if (_eventBuses.TryRemove(topic, out _) && eb.Handlers.Count > 0)
                    _eventBuses.TryAdd(topic, eb);
            }
        }
    }

    /// <summary>异步等待指定主题的第一条事件，超时后抛出 <see cref="OperationCanceledException"/></summary>
    /// <param name="topic">主题</param>
    /// <param name="timeout">超时时间</param>
    /// <returns>事件实例</returns>
    public async Task<TEvent> ReceiveAsync(String topic, TimeSpan timeout)
    {
        using var cts = new CancellationTokenSource(timeout);
        return await ReceiveAsync(topic, cts.Token).ConfigureAwait(false);
    }
    #endregion

    #region 日志
    /// <summary>日志</summary>
    public ILog Log { get; set; } = Logger.Null;

    /// <summary>写日志</summary>
    /// <param name="format">格式串</param>
    /// <param name="args">参数</param>
    public void WriteLog(String format, params Object[] args) => Log?.Info(format, args);
    #endregion
}