YD_TrainingCompetition_Api/Services/SerialIngestWorker.cs

using System.Threading.Channels;
using System.Net.Http;
using System.Text;
using System.Text.Json;
using Microsoft.Extensions.Options;
using YD_TrainingCompetition_Api.Competition;
using YD_TrainingCompetition_Api.Options;
using YD_TrainingCompetition_Api.Protocols;

namespace YD_TrainingCompetition_Api.Services;

public sealed class SerialIngestWorker : BackgroundService
{
    private static readonly HttpClient DebugClient = new();
    private readonly Serial.ISerialGateway _serialGateway;
    private readonly IJumpProtocolParser _parser;
    private readonly IDeviceStateStore _store;
    private readonly IRealtimeNotifier _notifier;
    private readonly ICompetitionSessionService _competition;
    private readonly IParticipantDirectory _directory;
    private readonly ILogger<SerialIngestWorker> _logger;
    private readonly Serial24GOptions _options;
    private readonly List<byte> _buffer = [];
    private readonly Channel<byte[]> _chunks = Channel.CreateUnbounded<byte[]>();

    public SerialIngestWorker(
        Serial.ISerialGateway serialGateway,
        IJumpProtocolParser parser,
        IDeviceStateStore store,
        IRealtimeNotifier notifier,
        ICompetitionSessionService competition,
        IParticipantDirectory directory,
        IOptions<Serial24GOptions> options,
        ILogger<SerialIngestWorker> logger)
    {
        _serialGateway = serialGateway;
        _parser = parser;
        _store = store;
        _notifier = notifier;
        _competition = competition;
        _directory = directory;
        _logger = logger;
        _options = options.Value;

        _serialGateway.DataReceived += (_, chunk) => _chunks.Writer.TryWrite(chunk);
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        try
        {
            await _serialGateway.OpenAsync(_options.PortName, stoppingToken);
            _logger.LogInformation("2.4G serial opened on {Port}", _options.PortName);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "Failed to open serial port {Port}. You can open it later from API.", _options.PortName);
        }

        await foreach (var chunk in _chunks.Reader.ReadAllAsync(stoppingToken))
        {
            await ProcessChunkAsync(chunk, stoppingToken);
        }
    }

    private async Task ProcessChunkAsync(byte[] chunk, CancellationToken cancellationToken)
    {
        #region debug-point C:chunk-received
        _ = DebugReportAsync("C", "SerialIngestWorker.ProcessChunkAsync", "[DEBUG] ingest chunk received", new
        {
            chunkLength = chunk.Length,
            bufferLength = _buffer.Count,
            hex = Convert.ToHexString(chunk)
        });
        #endregion
        _buffer.AddRange(chunk);

        while (_buffer.Count >= _options.FrameLength)
        {
            var start = _buffer.IndexOf(0xAA);
            if (start < 0)
            {
                _buffer.Clear();
                return;
            }

            if (_buffer.Count < start + _options.FrameLength)
            {
                if (start > 0)
                {
                    _buffer.RemoveRange(0, start);
                }

                return;
            }

            var frame = _buffer.Skip(start).Take(_options.FrameLength).ToArray();
            _buffer.RemoveRange(0, start + _options.FrameLength);

            if (!_parser.TryParse(frame, out var parsed) || parsed is null)
            {
                #region debug-point C:parse-failed
                _ = DebugReportAsync("C", "SerialIngestWorker.ProcessChunkAsync", "[DEBUG] frame parse failed", new
                {
                    frameLength = frame.Length,
                    expectedLength = _options.FrameLength,
                    hex = Convert.ToHexString(frame)
                });
                #endregion
                continue;
            }

            #region debug-point C:parse-succeeded
            _ = DebugReportAsync("C", "SerialIngestWorker.ProcessChunkAsync", "[DEBUG] frame parse succeeded", new
            {
                frameLength = frame.Length,
                hex = Convert.ToHexString(frame),
                deviceNo = parsed.DeviceNo,
                counter = parsed.Counter,
                battery = parsed.Battery,
                mode = parsed.Mode,
                remainingSeconds = parsed.RemainingSeconds
            });
            #endregion

            var deviceId = parsed.DeviceNo.ToString();
            var roster = _directory.FindByDeviceId(deviceId);
            var telemetry = new Models.DeviceTelemetry
            {
                DeviceId = deviceId,
                BsName = $"NO-{parsed.DeviceNo}",
                AthleteName = roster?.Name,
                School = roster?.School,
                Counter = parsed.Counter,
                ErrorCount = null,
                Battery = parsed.Battery,
                Mode = parsed.Mode,
                RemainingSeconds = parsed.RemainingSeconds,
                Timestamp = DateTimeOffset.UtcNow,
                IngestedAt = DateTimeOffset.UtcNow,
                GatewayId = "serial-2.4g"
            };

            if (!_store.TryUpsert(telemetry, out _))
            {
                continue;
            }

            var snapshot = _store.GetByDeviceId(deviceId);
            if (snapshot is not null)
            {
                await _notifier.BroadcastTelemetryAsync(snapshot, cancellationToken);
                await _competition.ApplyDeviceSnapshotAsync(snapshot, cancellationToken);
            }
        }
    }

    private static async Task DebugReportAsync(string hypothesisId, string location, string msg, object data)
    {
        try
        {
            var url = "http://127.0.0.1:7777/event";
            var sessionId = "serial-gateway-download";
            var envPath = Path.Combine(Directory.GetCurrentDirectory(), ".dbg", "serial-gateway-download.env");
            if (File.Exists(envPath))
            {
                foreach (var line in await File.ReadAllLinesAsync(envPath))
                {
                    if (line.StartsWith("DEBUG_SERVER_URL=", StringComparison.Ordinal))
                    {
                        url = line["DEBUG_SERVER_URL=".Length..].Trim();
                    }
                    else if (line.StartsWith("DEBUG_SESSION_ID=", StringComparison.Ordinal))
                    {
                        sessionId = line["DEBUG_SESSION_ID=".Length..].Trim();
                    }
                }
            }

            var payload = JsonSerializer.Serialize(new
            {
                sessionId,
                runId = "pre-fix",
                hypothesisId,
                location,
                msg,
                data,
                ts = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
            });
            using var content = new StringContent(payload, Encoding.UTF8, "application/json");
            await DebugClient.PostAsync(url, content);
        }
        catch
        {
        }
    }
}