using System.Text.Json;
using Fleet.Core;
using Fleet.Link;
using Fleet.Rpc;
using MoShiro.Measurement;
using MoShiro.Protocol;

namespace MoShiro.Agent;

/// <summary>Agent → Backend 的上行：事件和测试状态是尽力而为的通知（链路断开时丢弃；列块走发件箱，不会丢）。</summary>
public sealed class BackendUplink
{
    private readonly TimeProvider _time;
    private volatile FleetLink? _link;

    public BackendUplink(TimeProvider time)
    {
        _time = time;
    }

    public void Attach(FleetLink link) => _link = link;

    public bool IsConnected => _link?.CurrentSession is not null;

    public async ValueTask EmitEventAsync(string kind, string? detail = null, ConnectedProbe? probe = null)
    {
        RpcSession<LinkContext>? session = _link?.CurrentSession;
        if (session is null)
        {
            return;
        }

        var parameters = new AgentEventParams { Kind = kind, AtUtcMs = _time.GetUtcNow().ToUnixTimeMilliseconds(), Detail = detail, Probe = probe };
        await SafeAsync(() => session.NotifyAsync(MoShiroMethods.Event, parameters, MoShiroJsonContext.Default.AgentEventParams, CancellationToken.None));
    }

    public async ValueTask SendHostAsync(HostSnapshot snapshot)
    {
        RpcSession<LinkContext>? session = _link?.CurrentSession;
        if (session is null)
        {
            return;
        }

        await SafeAsync(() => session.NotifyAsync(MoShiroMethods.Host, snapshot, MoShiroJsonContext.Default.HostSnapshot, CancellationToken.None));
    }

    public async ValueTask SendTestStatusAsync(TestId test, string status, int? phase = null, string? errorCode = null)
    {
        RpcSession<LinkContext>? session = _link?.CurrentSession;
        if (session is null)
        {
            return;
        }

        var parameters = new TestStatusParams { TestId = test, Status = status, Phase = phase, ErrorCode = errorCode };
        await SafeAsync(() => session.NotifyAsync(MoShiroMethods.TestStatus, parameters, MoShiroJsonContext.Default.TestStatusParams, CancellationToken.None));
    }

    private static async ValueTask SafeAsync(Func<ValueTask> send)
    {
        try
        {
            await send();
        }
        catch (Exception ex) when (ex is ObjectDisposedException or InvalidOperationException or IOException or System.Net.WebSockets.WebSocketException)
        {
            // 链路刚好断开：状态在重连之后由 hello 与下一次事件补上
        }
    }
}

/// <summary>
/// Probe 的点对点会话：每次握手成功之后登记会话（同一个 Probe 的新会话取代它自己的旧会话，不同 Probe 的会话并存）、为这个会话起一组线路任务的流并下发线路任务集、接收 Probe 的列块并交给
/// <see cref="ChunkAssembler"/>。会话结束时通知 Backend，并让正在运行的测试以 <c>probe_offline</c> 结束。
/// </summary>
public sealed class PeerHub
{
    private const int DedupeWindow = 1024;

    /// <summary>超过这么久没有收到 Probe 的任何帧就认为会话已死（Probe 每 30 秒至少发一次时钟同步）。</summary>
    private static readonly TimeSpan IdleLimit = TimeSpan.FromSeconds(75);

        private readonly RpcRouter<PeerContext> _router;
    private readonly SessionRegistry _sessions;
    private readonly MonoClock _clock;
    private readonly ChunkAssembler _chunks;
    private readonly TaskManager _tasks;
    private readonly BackendUplink _uplink;
    private readonly TimeProvider _time;
    private readonly INodeLog _log;

    /// <summary>对等面入站方法的路由器（只读，测试用）。</summary>
    public RpcRouter<PeerContext> Router => _router;

    public PeerHub(SessionRegistry sessions, MonoClock clock, ChunkAssembler chunks, TaskManager tasks, BackendUplink uplink, TimeProvider time, INodeLog log)
    {
        _router = BuildRouter();
        _sessions = sessions;
        _clock = clock;
        _chunks = chunks;
        _tasks = tasks;
        _uplink = uplink;
        _time = time;
        _log = log;
    }

    /// <summary>Probe 会话断开（这个 Probe 上的测试据此中止）。</summary>
    public Action<ProbeSession>? OnProbeLost { get; set; }

    public async Task OnSessionAsync(PeerSessionInfo info, IRpcTransport transport, CancellationToken ct)
    {
        var probe = new ProbeSession(info) { ConnectedAtUtcMs = _time.GetUtcNow().ToUnixTimeMilliseconds() };
        var rpc = new RpcSession<PeerContext>(transport, _router, new PeerContext(probe, _clock), _time);
        probe.Rpc = rpc;

        var seen = new HashSet<Guid>();
        var order = new Queue<Guid>();
        rpc.OnBinary = async (frame, token) =>
        {
            if (!OutboxFrame.TryParse(frame, out byte channel, out Guid id, out ReadOnlyMemory<byte> payload) || channel != MoShiroChannels.ColumnBlock)
            {
                return;
            }

            // 至少一次投递：重复到达的列块只处理一次，但每次都要确认
            if (seen.Add(id))
            {
                order.Enqueue(id);
                if (order.Count > DedupeWindow)
                {
                    seen.Remove(order.Dequeue());
                }

                Result<RawChunk> decoded = RawChunkCodec.Decode(payload.Span);
                if (decoded.TryGet(out RawChunk? fragment, out FleetError? error))
                {
                    _chunks.Accept(fragment, probe);
                }
                else
                {
                    _log.Warn("fragment undecodable", ("err", error.Code), ("detail", error.Message));
                }
            }

            var ack = new OutboxAckParams { Ids = [id.ToString()] };
            await rpc.NotifyAsync(FleetMethods.OutboxAck, ack, FleetRpcContext.Default.OutboxAckParams, token);
        };

        ProbeSession? previous = _sessions.Replace(probe);
        if (previous?.Rpc is { } old)
        {
            await old.CloseAsync("replaced", CancellationToken.None);
        }

        // 这个会话自己的一组流：序号从 0 开始，上一个会话留下的记录不会算到它头上
        LineTaskSet initial = _tasks.OpenSession(probe);

        using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct);
        Task run = rpc.RunAsync(linked.Token);
        Task watchdog = WatchIdleAsync(probe, rpc, linked.Token);
        _log.Info("probe connected", ("probe", info.PeerId), ("version", info.Version ?? ""), ("remote", probe.RemoteAddress ?? ""));
        await _uplink.EmitEventAsync(AgentEventKinds.ProbeConnected, info.PeerId, Describe(probe));
        await PushAsync(probe, initial, linked.Token);
        try
        {
            await run;
        }
        catch (OperationCanceledException)
        {
            // 停止
        }
        finally
        {
            _sessions.Remove(probe);
            await linked.CancelAsync();
            await watchdog;
            await rpc.DisposeAsync();
            await _tasks.CloseSessionAsync(probe);
            _log.Info("probe disconnected", ("probe", info.PeerId));
            OnProbeLost?.Invoke(probe);
            await _uplink.EmitEventAsync(AgentEventKinds.ProbeDisconnected, info.PeerId, Describe(probe));
        }
    }

    /// <summary>
    /// Probe 每 30 秒发一次时钟同步请求，所以会话里至少每 30 秒有一帧：超过 <see cref="IdleLimit"/> 没有收到任何东西，
    /// 说明 Probe 已经静默死亡（断电、进程被冻结），关闭会话，让 Backend 看到 probe_disconnected，而不是一直显示在线。
    /// </summary>
    private async Task WatchIdleAsync(ProbeSession probe, RpcSession<PeerContext> rpc, CancellationToken ct)
    {
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(10), _time);
        try
        {
            while (await timer.WaitForNextTickAsync(ct))
            {
                if (rpc.IdleFor > IdleLimit)
                {
                    _log.Warn("probe session is silent, closing it", ("probe", probe.ProbeId), ("idle_s", (long)rpc.IdleFor.TotalSeconds));
                    await rpc.CloseAsync("idle", CancellationToken.None);
                    return;
                }
            }
        }
        catch (OperationCanceledException)
        {
            // 会话结束
        }
    }

    /// <summary>当前连着的 Probe（心跳和 Backend 重连之后的补报都用它）。</summary>
    public IReadOnlyList<ConnectedProbe> ConnectedProbes() => [.. _sessions.All.Select(Describe)];

    private static ConnectedProbe Describe(ProbeSession p) => new() { Id = new ProbeId(p.ProbeId), Version = p.Version, Remote = p.RemoteAddress, SinceUtcMs = p.ConnectedAtUtcMs, IpVersion = p.IpVersion };

    public IReadOnlyList<string> ConnectedProbeIds() => [.. _sessions.All.Select(p => p.ProbeId)];

    /// <summary>向一个 Probe 下发它的线路任务集（Probe 连上时由 <see cref="OnSessionAsync"/> 先发一份）。</summary>
    public async ValueTask PushToAsync(ProbeSession probe, LineTaskSet set) => await PushAsync(probe, set, CancellationToken.None);

    /// <summary>Backend 要求断开某个 Probe（它会按退避重连，除非已经被吊销）。没有这个会话时返回 false。</summary>
    public async ValueTask<bool> DisconnectAsync(string probeId)
    {
        bool disconnected = false;
        foreach (ProbeSession probe in _sessions.All)
        {
            if (probe.ProbeId == probeId && probe.Rpc is { } rpc)
            {
                await rpc.CloseAsync("disconnected", CancellationToken.None);
                disconnected = true;
            }
        }

        return disconnected;
    }

    /// <summary>名单（期望状态里的 peers）里已经没有的 Probe：立刻关闭它的会话（吊销生效）。</summary>
    public void CloseNotIn(IReadOnlySet<string> allowed)
    {
        foreach (ProbeSession probe in _sessions.All)
        {
            if (!allowed.Contains(probe.ProbeId) && probe.Rpc is { } rpc)
            {
                _log.Info("probe is no longer allowed, closing its session", ("probe", probe.ProbeId));
                _ = rpc.CloseAsync("revoked", CancellationToken.None).AsTask();
            }
        }
    }

    private async ValueTask PushAsync(ProbeSession probe, LineTaskSet set, CancellationToken ct)
    {
        if (probe.Rpc is null)
        {
            return;
        }

        var parameters = new StateSyncParams
        {
            Type = LineTaskSet.StateType,
            Version = set.Version,
            Doc = JsonSerializer.SerializeToElement(set, MoShiroJsonContext.Default.LineTaskSet),
        };
        try
        {
            Result<StateSyncResult> result = await probe.Rpc.RequestAsync(FleetMethods.StateSync, parameters, FleetRpcContext.Default.StateSyncParams, FleetRpcContext.Default.StateSyncResult, ct);
            if (!result.IsSuccess)
            {
                _log.Warn("probe rejected the line task set", ("err", result.Error.Code));
            }
        }
        catch (Exception ex) when (ex is ObjectDisposedException or InvalidOperationException or IOException or OperationCanceledException)
        {
            // 会话已经结束：下一个会话连上时补发
        }
    }

    private static RpcRouter<PeerContext> BuildRouter() =>
        new RpcRouter<PeerContext>(RpcSide.Listener, MoShiroMethodTable.Table)
            .On(
                MoShiroMethods.Clock,
                MoShiroJsonContext.Default.ClockParams,
                MoShiroJsonContext.Default.ClockResult,
                (context, _, _) => ValueTask.FromResult(Result<ClockResult>.Ok(new ClockResult { AgentMonoUs = context.Clock.NowUs() })));
}