YD_TrainingCompetition_Api/Services/MqttIngestWorker.cs

using System.Text;
using System.Text.Json;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Protocol;
using YD_TrainingCompetition_Api.Models;
using YD_TrainingCompetition_Api.Options;

namespace YD_TrainingCompetition_Api.Services;

public sealed class MqttIngestWorker : BackgroundService
{
    private readonly ILogger<MqttIngestWorker> _logger;
    private readonly IDeviceStateStore _store;
    private readonly IRealtimeNotifier _notifier;
    private readonly MqttOptions _options;

    public MqttIngestWorker(
        ILogger<MqttIngestWorker> logger,
        IDeviceStateStore store,
        IRealtimeNotifier notifier,
        IConfiguration configuration)
    {
        _logger = logger;
        _store = store;
        _notifier = notifier;
        _options = configuration.GetSection(MqttOptions.SectionName).Get<MqttOptions>() ?? new MqttOptions();
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var factory = new MqttFactory();
        using var client = factory.CreateMqttClient();

        client.ApplicationMessageReceivedAsync += async e => await HandleMessageAsync(e, stoppingToken);

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                var optionsBuilder = new MqttClientOptionsBuilder()
                    .WithClientId(_options.ClientId)
                    .WithTcpServer(_options.Host, _options.Port);

                if (!string.IsNullOrWhiteSpace(_options.Username))
                {
                    optionsBuilder.WithCredentials(_options.Username, _options.Password);
                }

                await client.ConnectAsync(optionsBuilder.Build(), stoppingToken);

                var topicFilter = new MqttTopicFilterBuilder()
                    .WithTopic(_options.TopicFilter)
                    .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce)
                    .Build();

                await client.SubscribeAsync(topicFilter, stoppingToken);
                _logger.LogInformation("MQTT ready: {Host}:{Port}, topic={Topic}", _options.Host, _options.Port, _options.TopicFilter);

                while (client.IsConnected && !stoppingToken.IsCancellationRequested)
                {
                    await Task.Delay(1000, stoppingToken);
                }
            }
            catch (OperationCanceledException)
            {
                break;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "MQTT worker failed, retrying in 2 seconds");
                await Task.Delay(2000, stoppingToken);
            }
        }
    }

    private async Task HandleMessageAsync(MqttApplicationMessageReceivedEventArgs e, CancellationToken cancellationToken)
    {
        try
        {
            var segment = e.ApplicationMessage.PayloadSegment;
            var payloadText = segment.Count > 0 && segment.Array is not null
                ? Encoding.UTF8.GetString(segment.Array, segment.Offset, segment.Count)
                : string.Empty;

            var telemetry = TryParseStandardTelemetry(payloadText, e.ApplicationMessage.Topic);
            if (telemetry is null)
            {
                return;
            }

            telemetry.IngestedAt = DateTimeOffset.UtcNow;

            if (!_store.TryUpsert(telemetry, out var reason))
            {
                _logger.LogWarning("Rejected telemetry {DeviceId}: {Reason}", telemetry.DeviceId, reason);
                return;
            }

            var snapshot = _store.GetByDeviceId(telemetry.DeviceId);
            if (snapshot is null)
            {
                return;
            }

            await _notifier.BroadcastTelemetryAsync(snapshot, cancellationToken);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to process MQTT message on topic {Topic}", e.ApplicationMessage.Topic);
        }
    }

    private static DeviceTelemetry? TryParseStandardTelemetry(string payloadText, string topic)
    {
        var telemetry = JsonSerializer.Deserialize<DeviceTelemetry>(payloadText, new JsonSerializerOptions
        {
            PropertyNameCaseInsensitive = true
        });

        if (telemetry is null || string.IsNullOrWhiteSpace(telemetry.DeviceId))
        {
            return null;
        }

        telemetry.Timestamp = telemetry.Timestamp == default ? DateTimeOffset.UtcNow : telemetry.Timestamp;
        telemetry.GatewayId ??= ParseGatewayId(topic);
        return telemetry;
    }

    private static string? ParseGatewayId(string topic)
    {
        var parts = topic.Split('/', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries);
        return parts.Length >= 2 ? parts[1] : null;
    }
}