using System.Net;
using System.Net.Http;
using System.Net.Quic;
using System.Net.Security;
using System.Net.Sockets;
using System.Security.Authentication;
using Fleet.Core;
using MoShiro.Protocol;

namespace MoShiro.Measurement;

/// <summary>http 请求的结果，编码与 Backend 的 <c>HttpOutcome</c> 一致。</summary>
internal static class HttpOutcomes
{
    public const int Ok = 0;
    public const int Timeout = 1;
    public const int Reset = 2;
    public const int Tls = 3;
    public const int Other = 4;
    public const int Dns = 5;
    public const int BadStatus = 6;
    public const int Unsupported = 7;
    public const int NotAllowed = 8;
}

/// <summary>解析出的地址都不在本地允许的网段内。</summary>
internal sealed class TargetNotAllowedException : Exception
{
    public TargetNotAllowedException(string host)
        : base("no address of " + host + " is allowed by allowed_target_cidrs")
    {
    }
}

/// <summary>
/// 单个 Agent 去请求一个 URL（HTTP/1.1、HTTP/2 或 HTTP/3，不回退），记录 DNS、TCP 建连、首字节、总耗时和状态码。
/// 每次请求新建连接，所以每次都包含完整的建连过程；频率按失败率自适应：目标一直应答就降到证明得了成功率的最低频率，
/// 出现失败升到最高频率，全部失败时指数退避。
/// </summary>
public sealed class HttpDialer : BlockBase
{
    private readonly DriverLog _log;
    private readonly int _attempt;
    private readonly int _ts;
    private readonly int _w;
    private readonly int _dns;
    private readonly int _tcp;
    private readonly int _ttfb;
    private readonly int _total;
    private readonly int _status;
    private readonly int _proto;
    private readonly int _outcome;
    private readonly HashSet<int> _loggedOutcomes = [];
    private long _next;

    public HttpDialer(BlockEnv env, BlockSpec spec)
        : base(env, spec)
    {
        _log = new DriverLog(spec.Kind.FindStream("req")!, Side.Dialer, 1024);
        _attempt = _log.Store.ColumnIndex("attempt");
        _ts = _log.Store.ColumnIndex("ts");
        _w = _log.Store.ColumnIndex("w");
        _dns = _log.Store.ColumnIndex("dns_us");
        _tcp = _log.Store.ColumnIndex("tcp_us");
        _ttfb = _log.Store.ColumnIndex("ttfb_us");
        _total = _log.Store.ColumnIndex("total_us");
        _status = _log.Store.ColumnIndex("status");
        _proto = _log.Store.ColumnIndex("proto");
        _outcome = _log.Store.ColumnIndex("outcome");
    }

    public override long Produced => _log.Count;

    public override async Task RunAsync(CancellationToken ct)
    {
        if (!TargetHost.TryParse(Env.Target, out _, out _, out Uri? url) || url is null)
        {
            Env.Log.Warn("http target is not a URL", ("task", Spec.Scope.Task?.Value));
            await WaitUntilCancelledAsync(ct);
            return;
        }

        AdaptiveRate? rate = CreateRequestRate();
        try
        {
            if (rate is not null)
            {
                await PaceAsync(rate, async () =>
                {
                    int weight = rate.Weight;
                    long sent = Env.Clock.NowUs();
                    rate.OnSend(sent);
                    bool ok = await RequestAsync(url, weight, ct);
                    if (ok)
                    {
                        long now = Env.Clock.NowUs();
                        rate.OnReply(sent, now, (now - sent) / 1000.0);
                    }
                }, ct);
            }
            else
            {
                TimeSpan interval = TimeSpan.FromSeconds(60.0 / Spec.Int("per_minute"));
                while (!ct.IsCancellationRequested)
                {
                    long start = Env.Clock.NowUs();
                    await RequestAsync(url, 1, ct);
                    TimeSpan wait = interval - TimeSpan.FromMicroseconds(Env.Clock.NowUs() - start);
                    if (wait > TimeSpan.Zero)
                    {
                        await Task.Delay(wait, Env.Time, ct);
                    }
                }
            }
        }
        catch (OperationCanceledException)
        {
            // 积木被停止
        }
    }

    public override int Flush(FlushMode mode, long peerProduced = 0)
    {
        RawStream? stream = _log.Take(null, all: true);
        if (stream is null)
        {
            return 0;
        }

        Ship(stream, "ts");
        return 1;
    }

    /// <summary>
    /// 频率以「次/分钟」声明，换算成 Hz 交给 <see cref="AdaptiveRate"/>。请求间隔以秒到分钟计，所以窗口按分钟级放宽，
    /// 并且不用 RTT 分位数的精度保护（样本太少，估计不出来）：平稳时只由成功率的上界决定频率。
    /// </summary>
    private AdaptiveRate? CreateRequestRate()
    {
        if (!Spec.Bool("adaptive"))
        {
            return null;
        }

        double max = Math.Max(Spec.Int("rate_max_per_min"), Spec.Int("rate_min_per_min")) / 60.0;
        ReportStart(max);
        return new AdaptiveRate(new AdaptiveRateOptions
        {
            MinHz = Spec.Int("rate_min_per_min") / 60.0,
            MaxHz = max,
            TauPct = Spec.Number("tau_pct"),
            RttGuard = false,
            CertifySeconds = 900,
            QuietCertifySeconds = 1200,
            GraceSeconds = Math.Ceiling(Spec.Int("timeout_ms") / 1000.0) + 1,
            DeadWindowSeconds = 120,
            DeadMinSent = 3,
            DeadMaxIntervalSeconds = 300,
            ActiveLossCount = 2,
            Prior = Env.LossPrior,
        });
    }

    /// <summary>发一次请求并记一行；返回这次算不算成功（有应答且状态码合格）。</summary>
    private async Task<bool> RequestAsync(Uri url, int weight, CancellationToken ct)
    {
        string version = Spec.Text("version");
        bool verify = Spec.Bool("tls_verify");
        TimeSpan timeout = TimeSpan.FromMilliseconds(Spec.Int("timeout_ms"));
        var phases = new Phases();
        long start = Env.Clock.NowUs();
        int outcome = HttpOutcomes.Other;
        int status = 0;
        int proto = 0;
        long? ttfb = null;
        long? total = null;
        using var timeoutCts = new CancellationTokenSource(timeout, Env.Time);
        using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct, timeoutCts.Token);
        try
        {
            if (version == "h3" && !QuicConnection.IsSupported)
            {
                outcome = HttpOutcomes.Unsupported;
                LogFirstFailure(outcome, url, version, null);
            }
            else
            {
                if (version == "h3")
                {
                    await ResolveForQuicAsync(url.IdnHost, phases, linked.Token);
                }

                using HttpMessageInvoker invoker = CreateInvoker(version, verify, phases);
                using var request = new HttpRequestMessage(new HttpMethod(Spec.Text("method")), url)
                {
                    Version = version switch { "h1" => HttpVersion.Version11, "h2" => HttpVersion.Version20, _ => HttpVersion.Version30 },
                    VersionPolicy = HttpVersionPolicy.RequestVersionExact,
                };
                using HttpResponseMessage response = await invoker.SendAsync(request, linked.Token);
                ttfb = Env.Clock.NowUs() - start;
                status = (int)response.StatusCode;
                proto = response.Version.Major switch { 3 => 3, 2 => 2, _ => 1 };
                await ReadBodyAsync(response, (int)Spec.Int("read_kb") * 1024, linked.Token);
                total = Env.Clock.NowUs() - start;
                long expected = Spec.Int("ok_status");
                outcome = (expected == 0 ? status < 400 : status == expected) ? HttpOutcomes.Ok : HttpOutcomes.BadStatus;
            }
        }
        catch (OperationCanceledException) when (!ct.IsCancellationRequested)
        {
            outcome = HttpOutcomes.Timeout;
        }
        catch (OperationCanceledException)
        {
            return false;
        }
        catch (Exception ex) when (ex is HttpRequestException or IOException or SocketException or AuthenticationException or TargetNotAllowedException or QuicException)
        {
            outcome = Classify(ex);
            LogFirstFailure(outcome, url, version, ex);
        }

        Record(start, weight, phases, ttfb, total, status, proto, outcome);
        return outcome == HttpOutcomes.Ok;
    }

    private HttpMessageInvoker CreateInvoker(string version, bool verify, Phases phases)
    {
        var handler = new SocketsHttpHandler
        {
            UseProxy = false,
            AllowAutoRedirect = false,
            UseCookies = false,
            AutomaticDecompression = DecompressionMethods.None,
        };

        // 设置了 ConnectCallback 的处理器不能用 HTTP/3（运行时用自己的 QUIC 连接，不经过回调）：h3 只能靠请求前的解析和网段检查
        if (version != "h3")
        {
            handler.ConnectCallback = (context, token) => ConnectAsync(context.DnsEndPoint, phases, token);
        }

        if (!verify)
        {
#pragma warning disable CA5359 // 操作员明确关闭证书校验（tls_verify=false）：只测可达性和耗时
            handler.SslOptions = new SslClientAuthenticationOptions { RemoteCertificateValidationCallback = (_, _, _, _) => true };
#pragma warning restore CA5359
        }

        return new HttpMessageInvoker(handler, disposeHandler: true);
    }

    /// <summary>自己解析和连接：每个地址都要在允许的网段里，DNS 和 TCP 建连的耗时分别记下来。</summary>
    private async ValueTask<Stream> ConnectAsync(DnsEndPoint endpoint, Phases phases, CancellationToken ct)
    {
        IPAddress[] addresses = await ResolveAsync(endpoint.Host, phases, ct);
        long start = Env.Clock.NowUs();
        Exception? last = null;
        foreach (IPAddress address in addresses)
        {
            var socket = new Socket(address.AddressFamily, SocketType.Stream, ProtocolType.Tcp) { NoDelay = true };
            try
            {
                await socket.ConnectAsync(new IPEndPoint(address, endpoint.Port), ct);
                phases.TcpUs = Env.Clock.NowUs() - start;
                return new NetworkStream(socket, ownsSocket: true);
            }
            catch (SocketException ex)
            {
                socket.Dispose();
                last = ex;
            }
        }

        throw last ?? new SocketException((int)SocketError.HostNotFound);
    }

    /// <summary>HTTP/3 由运行时自己建 UDP 连接，没法接管；先解析并检查一遍，再让它去连。</summary>
    private async Task ResolveForQuicAsync(string host, Phases phases, CancellationToken ct) => _ = await ResolveAsync(host, phases, ct);

    private async Task<IPAddress[]> ResolveAsync(string host, Phases phases, CancellationToken ct)
    {
        long start = Env.Clock.NowUs();
        IPAddress[] all = IPAddress.TryParse(host, out IPAddress? literal) ? [literal] : await Dns.GetHostAddressesAsync(host, ct);
        phases.DnsUs = Env.Clock.NowUs() - start;
        IPAddress[] allowed = [.. all.Where(a => Env.AllowAddress?.Invoke(a) ?? true)];
        return allowed.Length > 0 ? allowed : throw new TargetNotAllowedException(host);
    }

    private static async Task ReadBodyAsync(HttpResponseMessage response, int limit, CancellationToken ct)
    {
        if (limit <= 0)
        {
            return;
        }

        await using Stream body = await response.Content.ReadAsStreamAsync(ct);
        byte[] buffer = new byte[Math.Min(limit, 16 * 1024)];
        int read = 0;
        while (read < limit)
        {
            int n = await body.ReadAsync(buffer.AsMemory(0, Math.Min(buffer.Length, limit - read)), ct);
            if (n == 0)
            {
                break;
            }

            read += n;
        }
    }

    /// <summary>每种失败结果只记第一次（带原因），保留给事后审计，不会随请求频率刷屏。</summary>
    private void LogFirstFailure(int outcome, Uri url, string version, Exception? ex)
    {
        if (!_loggedOutcomes.Add(outcome))
        {
            return;
        }

        Env.Log.Warn("http request failed", ("task", Spec.Scope.Task?.Value), ("outcome", outcome), ("version", version), ("host", url.Host),
            ("quic_supported", version == "h3" ? QuicConnection.IsSupported : null), ("err", ex is null ? "System.Net.Quic is not available" : ex.GetType().Name + ": " + (ex.InnerException?.Message ?? ex.Message)));
    }

    private static int Classify(Exception ex)
    {
        for (Exception? e = ex; e is not null; e = e.InnerException)
        {
            switch (e)
            {
                case TargetNotAllowedException:
                    return HttpOutcomes.NotAllowed;
                case AuthenticationException:
                    return HttpOutcomes.Tls;
                case SocketException { SocketErrorCode: SocketError.HostNotFound or SocketError.TryAgain or SocketError.NoData }:
                    return HttpOutcomes.Dns;
                case SocketException { SocketErrorCode: SocketError.ConnectionRefused or SocketError.ConnectionReset }:
                    return HttpOutcomes.Reset;
                case HttpRequestException { HttpRequestError: HttpRequestError.NameResolutionError }:
                    return HttpOutcomes.Dns;
                case HttpRequestException { HttpRequestError: HttpRequestError.SecureConnectionError }:
                    return HttpOutcomes.Tls;
                case HttpRequestException { HttpRequestError: HttpRequestError.VersionNegotiationError }:
                    return HttpOutcomes.Unsupported;
            }
        }

        return HttpOutcomes.Other;
    }

    private void Record(long startUs, int weight, Phases phases, long? ttfb, long? total, int status, int proto, int outcome)
    {
        long key = _log.Store.Append();
        _log.Store.Set(key, _attempt, _next++);
        _log.Store.Set(key, _ts, startUs);
        _log.Store.Set(key, _w, Math.Clamp(weight, 1, 65535));
        _log.Store.Set(key, _outcome, outcome);
        SetIf(key, _dns, phases.DnsUs);
        SetIf(key, _tcp, phases.TcpUs);
        SetIf(key, _ttfb, ttfb);
        SetIf(key, _total, total);
        if (status != 0)
        {
            _log.Store.Set(key, _status, status);
            _log.Store.Set(key, _proto, proto);
        }
    }

    private void SetIf(long key, int column, long? value)
    {
        if (value is { } v)
        {
            _log.Store.Set(key, column, Math.Clamp(v, 0, uint.MaxValue));
        }
    }

    /// <summary>一次请求里 DNS 和 TCP 建连用了多久（连接回调里填）。</summary>
    private sealed class Phases
    {
        public long? DnsUs { get; set; }

        public long? TcpUs { get; set; }
    }
}