Migu2.0/MiGu.Server/Ota/WatchDogClient.cs

using System.Diagnostics;
using System.Net.Sockets;
using System.Text;
using System.Text.Json;
using Microsoft.Extensions.Options;

namespace MiGu.Server.Ota;

public sealed class WatchDogClient
{
    private readonly IHttpClientFactory _httpFactory;
    private readonly OtaOptions _opt;
    private readonly ILogger<WatchDogClient> _log;
    private readonly JsonSerializerOptions _json = new() { PropertyNameCaseInsensitive = true };

    public WatchDogClient(IHttpClientFactory httpFactory, IOptions<OtaOptions> opt, ILogger<WatchDogClient> log)
    {
        _httpFactory = httpFactory;
        _opt = opt.Value;
        _log = log;
    }

    private HttpClient CreateClient(int? timeoutMs = null)
    {
        var c = _httpFactory.CreateClient(nameof(WatchDogClient));
        c.Timeout = TimeSpan.FromMilliseconds(timeoutMs ?? _opt.RequestTimeoutMs);
        return c;
    }

    private string Base(string ip) => $"http://{ip}:{_opt.WatchDogPort}";

    public async Task<(bool Ok, OtaAppVersions? M, OtaAppVersions? D, OtaAppVersions? C, string? Error)> GetMdcInfoAsync(string ip, CancellationToken ct)
    {
        try
        {
            using var client = CreateClient();
            using var resp = await client.GetAsync($"{Base(ip)}/getMDCInfo", ct);
            if (!resp.IsSuccessStatusCode)
                return (false, null, null, null, $"HTTP {(int)resp.StatusCode}");
            var text = await resp.Content.ReadAsStringAsync(ct);
            using var doc = JsonDocument.Parse(text);
            var root = doc.RootElement;
            return (true, ParseApp(root, "Medulla"), ParseApp(root, "Detour"), ParseApp(root, "Clumsy"), null);
        }
        catch (Exception ex)
        {
            return (false, null, null, null, ex.Message);
        }
    }

    private static OtaAppVersions? ParseApp(JsonElement root, string name)
    {
        if (!root.TryGetProperty(name, out var app) && !root.TryGetProperty(name.ToLowerInvariant(), out app))
            return null;
        return new OtaAppVersions
        {
            Exe = ParseComp(app, "exe"),
            Dll = ParseComp(app, "dll"),
            Pdb = ParseComp(app, "pdb")
        };
    }

    private static OtaComponentVersion? ParseComp(JsonElement app, string key)
    {
        if (!app.TryGetProperty(key, out var c)) return null;
        string? ver = null;
        string? time = null;
        if (c.ValueKind == JsonValueKind.Object)
        {
            if (c.TryGetProperty("version", out var v))
                ver = v.ValueKind == JsonValueKind.String ? v.GetString() : v.ToString();
            if (c.TryGetProperty("time", out var t))
                time = t.GetString();
        }
        else if (c.ValueKind == JsonValueKind.String)
        {
            ver = c.GetString();
        }
        return new OtaComponentVersion { Version = ver, Time = time };
    }

    public async Task<int?> MeasureRttMsAsync(string ip, CancellationToken ct)
    {
        try
        {
            var sw = Stopwatch.StartNew();
            using var tcp = new TcpClient();
            using var reg = ct.Register(() => { try { tcp.Close(); } catch { /* ignore */ } });
            var connectTask = tcp.ConnectAsync(ip, _opt.WatchDogPort);
            var done = await Task.WhenAny(connectTask, Task.Delay(Math.Min(3000, _opt.RequestTimeoutMs), ct));
            if (done != connectTask || !tcp.Connected) return null;
            await connectTask;
            sw.Stop();
            return (int)sw.ElapsedMilliseconds;
        }
        catch
        {
            return null;
        }
    }

    public async Task TriggerPullAsync(string ip, string serverBaseUrl, string time, CancellationToken ct)
    {
        // 现网 WatchDog 忽略 server 查询参数，固定 POST 到 http://{config.serverIP}:8000/upload-mdcs/{key}。
        // serverBaseUrl 仅作日志/未来兼容；真正要通必须：车上 serverIP=本机局域网 IP，且本机监听 ReceivePort。
        using var client = CreateClient(_opt.UploadTimeoutMs);
        var url = $"{Base(ip)}/getmdcsexe?time={Uri.EscapeDataString(time)}&server={Uri.EscapeDataString(serverBaseUrl.TrimEnd('/'))}";
        try
        {
            using var resp = await client.GetAsync(url, ct);
            var body = (await resp.Content.ReadAsStringAsync(ct)).Trim();
            _log.LogInformation("getmdcsexe {Ip} -> {Code} body={Body} (WatchDog will POST to its config.serverIP:{Port}/upload-mdcs/*; expect receiver {Base})",
                ip, (int)resp.StatusCode, body.Length > 200 ? body[..200] : body, _opt.ReceivePort, serverBaseUrl);
            if (!resp.IsSuccessStatusCode)
                throw new InvalidOperationException($"WatchDog getmdcsexe HTTP {(int)resp.StatusCode}: {body}");
            if (body.Contains("请配置", StringComparison.Ordinal)
                || body.Equals("false", StringComparison.OrdinalIgnoreCase)
                || body.Equals("\"false\"", StringComparison.OrdinalIgnoreCase))
            {
                throw new InvalidOperationException(
                    "WatchDog 拒绝拉包或回传失败。请确认：1) 车已配置 Medulla/Detour/Clumsy 路径；" +
                    $"2) watch_dog.json 的 serverIP 指向本机局域网 IP（车将 POST 到 serverIP:{_opt.ReceivePort}/upload-mdcs/*）；" +
                    $"3) 本机已监听 :{_opt.ReceivePort}。WatchDog 返回：{body}");
            }
        }
        catch (InvalidOperationException)
        {
            throw;
        }
        catch (Exception ex)
        {
            _log.LogWarning(ex, "getmdcsexe failed {Ip}", ip);
            throw;
        }
    }

    public async Task UploadComponentAsync(string ip, string componentKey, string localPath, string fileName, int bandwidthKbps, CancellationToken ct)
    {
        var endpoint = OtaPathMap.WatchDogUpdatePath(componentKey)
            ?? throw new ArgumentException($"未知组件 {componentKey}");
        await UploadFileAsync($"{Base(ip)}/{endpoint}", localPath, fileName, bandwidthKbps, null, ct);
    }

    public async Task UploadCustomFileAsync(string ip, string localPath, string fileName, string remotePath, int restartOp, int bandwidthKbps, CancellationToken ct)
    {
        var url = $"{Base(ip)}/updateFile/{Uri.EscapeDataString(fileName)}/{restartOp}/";
        await UploadFileAsync(url, localPath, fileName, bandwidthKbps, new Dictionary<string, string> { ["path"] = remotePath }, ct);
    }

    private async Task UploadFileAsync(string url, string localPath, string fileName, int bandwidthKbps, Dictionary<string, string>? extraFields, CancellationToken ct)
    {
        using var client = CreateClient(_opt.UploadTimeoutMs);
        await using var fs = File.OpenRead(localPath);
        Stream contentStream = fs;
        if (bandwidthKbps > 0)
            contentStream = new ThrottledStream(fs, bandwidthKbps * 1024L);

        using var form = new MultipartFormDataContent();
        if (extraFields != null)
        {
            foreach (var (k, v) in extraFields)
                form.Add(new StringContent(v, Encoding.UTF8), k);
        }
        var streamContent = new StreamContent(contentStream);
        form.Add(streamContent, "file", fileName);

        using var resp = await client.PostAsync(url, form, ct);
        if (!resp.IsSuccessStatusCode)
        {
            var body = await resp.Content.ReadAsStringAsync(ct);
            throw new InvalidOperationException($"上传失败 HTTP {(int)resp.StatusCode}: {body}");
        }
    }

    public async Task<string> GetJsonAsync(string ip, string app, CancellationToken ct)
    {
        var path = app.ToLowerInvariant() switch
        {
            "medulla" => "getMedullajson",
            "detour" => "getDetourjson",
            "clumsy" => "getClumsyjson",
            _ => throw new ArgumentException("app 须为 medulla|detour|clumsy")
        };
        using var client = CreateClient();
        using var resp = await client.GetAsync($"{Base(ip)}/{path}", ct);
        resp.EnsureSuccessStatusCode();
        return await resp.Content.ReadAsStringAsync(ct);
    }

    public async Task PutJsonAsync(string ip, string app, string json, CancellationToken ct)
    {
        var path = app.ToLowerInvariant() switch
        {
            "medulla" => "updateMedullajson",
            "detour" => "updateDetourjson",
            "clumsy" => "updateClumsyjson",
            _ => throw new ArgumentException("app 须为 medulla|detour|clumsy")
        };
        using var client = CreateClient();
        using var content = new StringContent(json, Encoding.UTF8, "application/json");
        using var resp = await client.PostAsync($"{Base(ip)}/{path}", content, ct);
        if (!resp.IsSuccessStatusCode)
        {
            var body = await resp.Content.ReadAsStringAsync(ct);
            throw new InvalidOperationException($"更新 JSON 失败 HTTP {(int)resp.StatusCode}: {body}");
        }
    }

    /// <summary>简易限速流：按字节/秒节流读取。</summary>
    private sealed class ThrottledStream : Stream
    {
        private readonly Stream _inner;
        private readonly long _bytesPerSecond;
        private long _windowBytes;
        private long _windowStart = Environment.TickCount64;

        public ThrottledStream(Stream inner, long bytesPerSecond)
        {
            _inner = inner;
            _bytesPerSecond = Math.Max(1024, bytesPerSecond);
        }

        public override bool CanRead => _inner.CanRead;
        public override bool CanSeek => false;
        public override bool CanWrite => false;
        public override long Length => _inner.Length;
        public override long Position { get => _inner.Position; set => throw new NotSupportedException(); }
        public override void Flush() => _inner.Flush();
        public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
        public override void SetLength(long value) => throw new NotSupportedException();
        public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();

        public override int Read(byte[] buffer, int offset, int count)
        {
            var n = _inner.Read(buffer, offset, count);
            if (n > 0) Throttle(n);
            return n;
        }

        public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
        {
            var n = await _inner.ReadAsync(buffer.AsMemory(offset, count), cancellationToken);
            if (n > 0) await ThrottleAsync(n, cancellationToken);
            return n;
        }

        private void Throttle(int n)
        {
            _windowBytes += n;
            var elapsed = Environment.TickCount64 - _windowStart;
            if (elapsed < 1) elapsed = 1;
            var allowed = _bytesPerSecond * elapsed / 1000;
            if (_windowBytes > allowed)
            {
                var wait = (int)((_windowBytes - allowed) * 1000 / _bytesPerSecond);
                if (wait > 0) Thread.Sleep(Math.Min(wait, 2000));
            }
            if (elapsed >= 1000)
            {
                _windowBytes = 0;
                _windowStart = Environment.TickCount64;
            }
        }

        private async Task ThrottleAsync(int n, CancellationToken ct)
        {
            _windowBytes += n;
            var elapsed = Environment.TickCount64 - _windowStart;
            if (elapsed < 1) elapsed = 1;
            var allowed = _bytesPerSecond * elapsed / 1000;
            if (_windowBytes > allowed)
            {
                var wait = (int)((_windowBytes - allowed) * 1000 / _bytesPerSecond);
                if (wait > 0) await Task.Delay(Math.Min(wait, 2000), ct);
            }
            if (elapsed >= 1000)
            {
                _windowBytes = 0;
                _windowStart = Environment.TickCount64;
            }
        }

        protected override void Dispose(bool disposing)
        {
            // 不释放 inner（由调用方 using FileStream）
            base.Dispose(disposing);
        }
    }
}