Deal/src/telegram-service/Deal.Telegram/Dialogs/BackfillService.cs

using System.Collections.Concurrent;
using Deal.Grpc.Sources;
using Deal.Telegram.Core;
using Deal.Telegram.Sessions;
using Deal.Telegram.Telegram;

namespace Deal.Telegram.Dialogs;

/// <summary>
/// Backfill диалога
/// </summary>
public sealed class BackfillService
{
    /// <summary>
    /// Сколько последних сообщений читает backfill.
    /// </summary>
    public const int MessagesLimit = 10;

    /// <summary>
    /// Нижняя граница анти-бан-паузы между сообщениями.
    /// </summary>
    public const double MinPerMessageDelaySeconds = 1.5;

    /// <summary>
    /// Верхняя граница анти-бан-паузы между сообщениями.
    /// </summary>
    public const double MaxPerMessageDelaySeconds = 3.0;

    /// <summary>
    /// Нижняя граница паузы между диалогами одного тенанта.
    /// </summary>
    public const double MinPerDialogDelaySeconds = 3.0;

    /// <summary>
    /// Верхняя граница паузы между диалогами одного тенанта.
    /// </summary>
    public const double MaxPerDialogDelaySeconds = 6.0;

    private readonly SessionFarm _sessionFarm;
    private readonly ICoreIngressClient _ingress;
    private readonly IBackfillPacer _pacer;
    private readonly ILogger<BackfillService> _logger;

    private readonly ConcurrentDictionary<(string TenantId, string DialogId), byte> _running = new();

    private readonly ConcurrentDictionary<string, SemaphoreSlim> _tenantGates = new(StringComparer.Ordinal);

    // Момент завершения последнего backfill тенанта (пауза 3–6 с между диалогами).
    private readonly Dictionary<string, DateTime> _tenantLastFinishUtc = new(StringComparer.Ordinal);

    private readonly object _lastFinishGate = new();

    /// <summary>
    /// Создаёт службу backfill'а.
    /// </summary>
    /// <param name="sessionFarm">Пул сессий тенантов (read-операции диалогов).</param>
    /// <param name="ingress">Канал в ядро (PushSource).</param>
    /// <param name="pacer">Анти-бан-паузы (реальный — случайные, тесты — фейк).</param>
    /// <param name="logger">Логгер.</param>
    public BackfillService(
        SessionFarm sessionFarm,
        ICoreIngressClient ingress,
        IBackfillPacer pacer,
        ILogger<BackfillService> logger)
    {
        _sessionFarm = sessionFarm;
        _ingress = ingress;
        _pacer = pacer;
        _logger = logger;
    }

    /// <summary>
    /// Перечитывает последние сообщения диалога в ядро
    /// </summary>
    /// <param name="tenantId">Id тенанта.</param>
    /// <param name="dialogId">Подписанный id диалога.</param>
    /// <param name="force">True — «Перечитать» по кнопке (семантика флага — в ядре, см. класс).</param>
    /// <param name="cancellationToken">Отмена операции.</param>
    /// <returns>Сколько сообщений отправлено в ядро (0 — диалог уже перечитывается/нет текстов).</returns>
    public async Task<int> ExecuteAsync(
        string tenantId,
        string dialogId,
        bool force,
        CancellationToken cancellationToken)
    {
        var key = (tenantId, dialogId);
        if (!_running.TryAdd(key, 0))
        {
            _logger.LogInformation("Аудит: backfill {TenantId} {DialogId} → пропущен (уже перечитывается)", tenantId, dialogId);
            return 0;
        }

        try
        {
            SemaphoreSlim gate = _tenantGates.GetOrAdd(tenantId, static _ => new SemaphoreSlim(1, 1));
            await gate.WaitAsync(cancellationToken).ConfigureAwait(false);
            try
            {
                await EnsureDialogSpacingAsync(tenantId, cancellationToken).ConfigureAwait(false);
                int processed = await BackfillDialogAsync(tenantId, dialogId, cancellationToken).ConfigureAwait(false);
                lock (_lastFinishGate)
                {
                    _tenantLastFinishUtc[tenantId] = DateTime.UtcNow;
                }

                _logger.LogInformation(
                    "Аудит: backfill {TenantId} {DialogId} → {Processed} сообщений в ядро (force {Force})",
                    tenantId,
                    dialogId,
                    processed,
                    force);
                return processed;
            }
            finally
            {
                gate.Release();
            }
        }
        finally
        {
            _running.TryRemove(key, out _);
        }
    }

    private async Task EnsureDialogSpacingAsync(string tenantId, CancellationToken cancellationToken)
    {
        DateTime? lastFinishUtc;
        lock (_lastFinishGate)
        {
            lastFinishUtc = _tenantLastFinishUtc.TryGetValue(tenantId, out DateTime value) ? value : null;
        }

        if (lastFinishUtc is null)
        {
            return;
        }

        double elapsedSeconds = (DateTime.UtcNow - lastFinishUtc.Value).TotalSeconds;
        if (elapsedSeconds >= MaxPerDialogDelaySeconds)
        {
            return;
        }

        double minSeconds = Math.Max(0, MinPerDialogDelaySeconds - elapsedSeconds);
        double maxSeconds = MaxPerDialogDelaySeconds - elapsedSeconds;
        if (maxSeconds < minSeconds)
        {
            maxSeconds = minSeconds;
        }

        await _pacer.WaitAsync(minSeconds, maxSeconds, cancellationToken).ConfigureAwait(false);
    }

    // Читает и отправляет сообщения одного диалога (паузы между сообщениями, read-ack).
    // tenantId: Id тенанта.
    // dialogId: Подписанный id диалога.
    // cancellationToken: Отмена операции.
    // Возвращает: Сколько сообщений отправлено в ядро.
    private async Task<int> BackfillDialogAsync(
        string tenantId,
        string dialogId,
        CancellationToken cancellationToken)
    {
        IReadOnlyList<TelegramMessage> messages = await _sessionFarm
            .GetMessagesAsync(tenantId, dialogId, MessagesLimit, cancellationToken)
            .ConfigureAwait(false);

        int processed = 0;
        foreach (TelegramMessage message in messages.Reverse())
        {
            PushSourceReply reply = await _ingress
                .PushSourceAsync(tenantId, DialogProtoMapper.ToSourceRequest(message), cancellationToken)
                .ConfigureAwait(false);
            if (reply.Accepted && !reply.Duplicate)
            {
                processed++;
            }

            await _pacer.WaitAsync(MinPerMessageDelaySeconds, MaxPerMessageDelaySeconds, cancellationToken).ConfigureAwait(false);
        }

        await _sessionFarm.MarkReadAsync(tenantId, dialogId, cancellationToken).ConfigureAwait(false);
        return processed;
    }
}