zengzy / hhrcs_service · Files

using System.Collections.Concurrent;
using MassTransit.Mediator;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Rcs.Application.Common;
using Rcs.Application.MessageBus.Commands;
using Rcs.Application.Services;
using Rcs.Domain.Entities;
using Rcs.Domain.Repositories;
using Rcs.Shared.Utils;
using TaskStatus = Rcs.Domain.Entities.TaskStatus;

namespace Rcs.Infrastructure.Services;

/// <summary>
/// Fatigue test background service. One worker per running fatigue config.
/// </summary>
public class FatigueTestBackgroundService : BackgroundService
{
    private readonly ILogger<FatigueTestBackgroundService> _logger;
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly TimeSpan _checkInterval = TimeSpan.FromSeconds(5);
    private readonly ConcurrentDictionary<string, WorkerState> _workers = new(StringComparer.OrdinalIgnoreCase);

    public FatigueTestBackgroundService(
        ILogger<FatigueTestBackgroundService> logger,
        IServiceScopeFactory scopeFactory)
    {
        _logger = logger;
        _scopeFactory = scopeFactory;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("[FatigueTest] Fatigue test background service started");

        try
        {
            while (!stoppingToken.IsCancellationRequested)
            {
                await SyncWorkersAsync(stoppingToken);
                await Task.Delay(_checkInterval, stoppingToken);
            }
        }
        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
        {
            // ignored
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "[FatigueTest] Supervisor loop crashed");
        }
        finally
        {
            await StopAllWorkersAsync();
            _logger.LogInformation("[FatigueTest] Fatigue test background service stopped");
        }
    }

    private async Task SyncWorkersAsync(CancellationToken cancellationToken)
    {
        await CleanupCompletedWorkersAsync();

        using var scope = _scopeFactory.CreateScope();
        var configService = scope.ServiceProvider.GetRequiredService<IFatigueTestConfigService>();
        var configs = await configService.GetConfigsAsync(cancellationToken);

        var runningConfigIds = configs
            .Where(c => c.IsRunning && !string.IsNullOrWhiteSpace(c.ConfigId))
            .Select(c => c.ConfigId)
            .ToHashSet(StringComparer.OrdinalIgnoreCase);

        var existingWorkerIds = _workers.Keys.ToList();
        foreach (var workerId in existingWorkerIds)
        {
            if (!runningConfigIds.Contains(workerId))
            {
                await StopWorkerAsync(workerId);
            }
        }

        foreach (var configId in runningConfigIds)
        {
            if (_workers.ContainsKey(configId))
            {
                continue;
            }

            var workerCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            var workerTask = RunWorkerAsync(configId, workerCts.Token);
            var workerState = new WorkerState(workerCts, workerTask);

            if (!_workers.TryAdd(configId, workerState))
            {
                workerCts.Cancel();
                workerCts.Dispose();
                continue;
            }

            _logger.LogInformation("[FatigueTest] Worker started for config {ConfigId}", configId);
        }
    }

    private async Task CleanupCompletedWorkersAsync()
    {
        foreach (var kvp in _workers.ToArray())
        {
            if (!kvp.Value.Task.IsCompleted)
            {
                continue;
            }

            if (!_workers.TryRemove(kvp.Key, out var removed))
            {
                continue;
            }

            try
            {
                await removed.Task;
            }
            catch (OperationCanceledException)
            {
                // ignored
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "[FatigueTest] Worker {ConfigId} failed", kvp.Key);
            }
            finally
            {
                removed.Cts.Dispose();
            }
        }
    }

    private async Task StopWorkerAsync(string configId)
    {
        if (!_workers.TryRemove(configId, out var worker))
        {
            return;
        }

        worker.Cts.Cancel();
        try
        {
            await worker.Task;
        }
        catch (OperationCanceledException)
        {
            // ignored
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "[FatigueTest] Worker {ConfigId} stopped with error", configId);
        }
        finally
        {
            worker.Cts.Dispose();
        }

        _logger.LogInformation("[FatigueTest] Worker stopped for config {ConfigId}", configId);
    }

    private async Task StopAllWorkersAsync()
    {
        var allWorkerIds = _workers.Keys.ToList();
        foreach (var configId in allWorkerIds)
        {
            await StopWorkerAsync(configId);
        }
    }

    private async Task RunWorkerAsync(string configId, CancellationToken cancellationToken)
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            try
            {
                using var scope = _scopeFactory.CreateScope();
                var configService = scope.ServiceProvider.GetRequiredService<IFatigueTestConfigService>();
                var config = await configService.GetConfigByIdAsync(configId, cancellationToken);

                if (config?.IsRunning != true)
                {
                    _logger.LogInformation("[FatigueTest] Worker exits for config {ConfigId} because config is stopped or deleted", configId);
                    return;
                }

                await ProcessSingleConfigOnceAsync(config, scope.ServiceProvider, cancellationToken);
            }
            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
            {
                return;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "[FatigueTest] Error while processing config {ConfigId}", configId);
            }

            await Task.Delay(_checkInterval, cancellationToken);
        }
    }

    private async Task ProcessSingleConfigOnceAsync(
        FatigueTestConfig config,
        IServiceProvider serviceProvider,
        CancellationToken cancellationToken)
    {
        if (!ValidateConfig(config))
        {
            return;
        }

        var robotRepository = serviceProvider.GetRequiredService<IRobotRepository>();
        var robotTaskRepository = serviceProvider.GetRequiredService<IRobotTaskRepository>();
        var storageLocationRepository = serviceProvider.GetRequiredService<IStorageLocationRepository>();
        var mediator = serviceProvider.GetRequiredService<IMediator>();

        var availableRobotAssignments = await GetAvailableRobotsAsync(
            config.RobotIds,
            robotRepository,
            robotTaskRepository,
            cancellationToken);

        if (!availableRobotAssignments.Any())
        {
            _logger.LogDebug("[FatigueTest] Config {ConfigId}: no available robot", config.ConfigId);
            return;
        }

        var locationIds = config.LocationIds
            .Where(id => Guid.TryParse(id, out _))
            .Select(Guid.Parse)
            .ToList();

        if (locationIds.Count < 2)
        {
            _logger.LogWarning("[FatigueTest] Config {ConfigId}: locations are less than 2", config.ConfigId);
            return;
        }

        var allLocations = new List<StorageLocation>();
        foreach (var locationId in locationIds)
        {
            var location = await storageLocationRepository.GetByIdAsync(locationId, cancellationToken);
            if (location != null && location.IsActive)
            {
                allLocations.Add(location);
            }
        }

        var occupiedLocations = allLocations
            .Where(l => l.Status == StorageLocationStatus.Occupied)
            .ToList();
        var emptyLocations = allLocations
            .Where(l => l.Status == StorageLocationStatus.Empty)
            .ToList();

        if (!occupiedLocations.Any() || !emptyLocations.Any())
        {
            _logger.LogDebug(
                "[FatigueTest] Config {ConfigId}: insufficient occupied or empty locations, occupied={OccupiedCount}, empty={EmptyCount}",
                config.ConfigId,
                occupiedLocations.Count,
                emptyLocations.Count);
            return;
        }

        var random = Random.Shared;
        var (robot, cacheLocation) = availableRobotAssignments[random.Next(availableRobotAssignments.Count)];
        var beginLocation = occupiedLocations[random.Next(occupiedLocations.Count)];
        var endLocation = emptyLocations[random.Next(emptyLocations.Count)];

        if (beginLocation.LocationId == endLocation.LocationId)
        {
            _logger.LogDebug("[FatigueTest] Config {ConfigId}: begin and end locations are the same", config.ConfigId);
            return;
        }

        var hasActiveTask = await CheckLocationHasActiveTaskAsync(
            beginLocation.LocationId,
            endLocation.LocationId,
            robotTaskRepository,
            cancellationToken);

        if (hasActiveTask)
        {
            _logger.LogDebug("[FatigueTest] Config {ConfigId}: begin or end location already has active task", config.ConfigId);
            return;
        }

        await CreateTransportTaskAsync(
            robot,
            cacheLocation,
            beginLocation,
            endLocation,
            mediator,
            robotRepository,
            cancellationToken);

        _logger.LogInformation(
            "[FatigueTest] Config {ConfigId}: task created, robot={RobotCode}, begin={BeginLocation}, end={EndLocation}",
            config.ConfigId,
            robot.RobotCode,
            beginLocation.LocationCode,
            endLocation.LocationCode);

        await Task.Delay(config.TaskIntervalMs, cancellationToken);
    }

    private bool ValidateConfig(FatigueTestConfig config)
    {
        if (!config.RobotIds.Any())
        {
            _logger.LogWarning("[FatigueTest] Config {ConfigId} invalid: RobotIds is empty", config.ConfigId);
            return false;
        }

        if (!config.LocationIds.Any())
        {
            _logger.LogWarning("[FatigueTest] Config {ConfigId} invalid: LocationIds is empty", config.ConfigId);
            return false;
        }

        return true;
    }

    private async Task<List<(Robot Robot, RobotCacheLocation CacheLocation)>> GetAvailableRobotsAsync(
        List<string> robotIds,
        IRobotRepository robotRepository,
        IRobotTaskRepository robotTaskRepository,
        CancellationToken cancellationToken)
    {
        var availableRobots = new List<(Robot Robot, RobotCacheLocation CacheLocation)>();

        foreach (var robotIdStr in robotIds)
        {
            if (!Guid.TryParse(robotIdStr, out var robotId))
            {
                continue;
            }

            var robot = await robotRepository.GetByIdFullDataAsync(robotId, cancellationToken);
            if (robot?.Active != true)
            {
                continue;
            }

            var inProgressTasks = await robotTaskRepository.GetnProgressByRobotIdAsync(robotId, cancellationToken);
            if (inProgressTasks.Any())
            {
                _logger.LogDebug("[FatigueTest] Robot {RobotCode} has in-progress task, skipped", robot.RobotCode);
                continue;
            }

            var availableCacheLocation = robot.CacheLocations
                .FirstOrDefault(location => string.IsNullOrWhiteSpace(location.ContainerId));
            if (availableCacheLocation == null)
            {
                _logger.LogDebug("[FatigueTest] Robot {RobotCode} has no empty cache location, skipped", robot.RobotCode);
                continue;
            }

            availableRobots.Add((robot, availableCacheLocation));
        }

        return availableRobots;
    }

    private async Task<bool> CheckLocationHasActiveTaskAsync(
        Guid beginLocationId,
        Guid endLocationId,
        IRobotTaskRepository robotTaskRepository,
        CancellationToken cancellationToken)
    {
        var allTasks = await robotTaskRepository.GetAllAsync(cancellationToken);
        var activeTasks = allTasks.Where(t => t.Status != TaskStatus.Cancelled && t.Status != TaskStatus.Completed).ToList();

        foreach (var task in activeTasks)
        {
            if (task.BeginLocationId.HasValue &&
                (task.BeginLocationId.Value == beginLocationId || task.BeginLocationId.Value == endLocationId))
            {
                return true;
            }

            if (task.EndLocationId.HasValue &&
                (task.EndLocationId.Value == beginLocationId || task.EndLocationId.Value == endLocationId))
            {
                return true;
            }
        }

        return false;
    }

    private async Task CreateTransportTaskAsync(
        Robot robot,
        RobotCacheLocation cacheLocation,
        StorageLocation beginLocation,
        StorageLocation endLocation,
        IMediator mediator,
        IRobotRepository robotRepository,
        CancellationToken cancellationToken)
    {
        var command = new CreateOrUpdateRobotTaskCommand
        {
            TaskCode = $"FATIGUE_{DateTime.Now:yyMMddHHmmss}_{Guid.NewGuid().ToString("N")[..6].ToUpper()}",
            TaskName = $"FatigueTask_{DateTime.Now:HHmmss}",
            RobotId = robot.RobotId.ToString(),
            ShelfCode = cacheLocation.LocationCode,
            BeginLocationId = beginLocation.LocationId.ToString(),
            EndLocationId = endLocation.LocationId.ToString(),
            ContainerID = GenerateHash.GenerateHashLength8(),
            Priority = 50,
            Status = (int)TaskStatus.Pending
        };

        var client = mediator.CreateRequestClient<CreateOrUpdateRobotTaskCommand>();
        var response = await client.GetResponse<ApiResponse>(command, cancellationToken);

        if (!response.Message.Success)
        {
            throw new InvalidOperationException($"Create task failed: {response.Message.Message}");
        }

        var targetCacheLocation = robot.CacheLocations
            .FirstOrDefault(c => c.LocationCode == command.ShelfCode)
            ?? cacheLocation;

        targetCacheLocation.ContainerId = command.ContainerID;
        targetCacheLocation.UpdatedAt = DateTime.Now;

        await robotRepository.UpdateAsync(robot, cancellationToken);
        await robotRepository.SaveChangesAsync(cancellationToken);
    }

    private sealed class WorkerState
    {
        public WorkerState(CancellationTokenSource cts, Task task)
        {
            Cts = cts;
            Task = task;
        }

        public CancellationTokenSource Cts { get; }

        public Task Task { get; }
    }
}