Deal/src/core/Deal.Api/Telegram/TelegramIngressService.cs

using System.Text.Json;
using Deal.Api.Events;
using Deal.Contracts.Integrations.Models;
using Deal.Grpc.Telegram;
using Deal.Modules.Pipeline.Application.Models;
using Deal.Modules.Pipeline.Application.Services;
using Deal.Modules.Settings.Application.Abstractions;
using Deal.Modules.Settings.Application.Models;
using Deal.Modules.Telegram.Application;
using Deal.Modules.Tenants.Application.Abstractions;
using Deal.Modules.Tenants.Application.Models;
using Deal.SharedKernel.Tenants.Abstractions;
using Deal.SharedKernel.Tenants.Models;
using Grpc.Core;

namespace Deal.Api.Telegram;

/// <summary>
/// gRPC-сервер входящего потока telegram-service → ядро.
/// </summary>
public sealed class TelegramIngressService(
    IServiceScopeFactory scopeFactory,
    SseBroker broker,
    ILogger<TelegramIngressService> logger) : IngressService.IngressServiceBase
{
    public const string TenantIdMetadataKey = "tenant-id";

    private const string SystemStatusEventType = "system_status";

    private const string ToastEventType = "toast";

    private const string ConnectedToastText = "Telegram подключён, сессия сохранена";

    private const string DisconnectedToastText = "Telegram отключён";

    // Иконка тоста подключения (из набора Icon.vue фронта).
    private const string ConnectedToastIcon = "send";

    // Иконка тоста отключения (из набора Icon.vue фронта).
    private const string DisconnectedToastIcon = "logout";

    // Деталь отказа: metadata tenant-id отсутствует (UNAUTHENTICATED, README src/contracts).
    private const string MissingTenantIdDetail = "tenant-id отсутствует в metadata";

    private static readonly JsonSerializerOptions StatusJsonOptions = new()
    {
        PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
        PropertyNameCaseInsensitive = true,
    };

    /// <summary>
    /// PushMessage — сообщение диалога в очередь пайплайна тенанта + превью.
    /// </summary>
    /// <param name="request">Сообщение из потока telegram-service.</param>
    /// <param name="context">Контекст вызова (metadata tenant-id + service-token).</param>
    /// <returns>accepted — сообщение принято (либо дубль), duplicate — уже было в очереди.</returns>
    public override async Task<PushMessageReply> PushMessage(PushMessageRequest request, ServerCallContext context)
    {
        TenantRecordDto? tenant = await ResolveTenantAsync(context).ConfigureAwait(false);
        if (tenant is null)
        {
            return new PushMessageReply();
        }

        await using AsyncServiceScope tenantScope = scopeFactory.CreateAsyncScope();
        ITenantContext tenantContext = tenantScope.ServiceProvider.GetRequiredService<ITenantContext>();
        try
        {
            // Resolve ПОСЛЕ SetTenant: TenantDbContext (и его адаптеры) строятся от схемы текущего тенанта.
            tenantContext.SetTenant(new TenantId(tenant.Id.ToString("N")));
            PipelineIngestService ingest = tenantScope.ServiceProvider.GetRequiredService<PipelineIngestService>();

            PipelineIngestResultDto result = await ingest.EnqueueAsync(
                new QueuedMessage
                {
                    DialogId = request.DialogId,
                    ChannelName = request.ChannelName,
                    ChannelHandle = request.ChannelHandle,
                    ChannelHue = request.ChannelHue,
                    MsgId = request.HasMsgId ? request.MsgId : null,
                    Text = request.Text,
                    MsgAtMs = request.HasMsgAt ? request.MsgAt : null,
                },
                context.CancellationToken).ConfigureAwait(false);

            await SavePreviewSafelyAsync(tenantScope, tenant, request, context.CancellationToken).ConfigureAwait(false);

            logger.LogInformation(
                "Аудит: PushMessage {TenantId} диалог {DialogId} msg {MsgId} → {Outcome}",
                tenant.Id,
                request.DialogId,
                request.HasMsgId ? request.MsgId.ToString() : "-",
                result.Duplicate ? "duplicate" : result.Id is null ? "no-op" : "queued");

            return new PushMessageReply
            {
                Accepted = result.Id is not null || result.Duplicate,
                Duplicate = result.Duplicate,
            };
        }
        catch (OperationCanceledException)
        {
            throw;
        }
        catch (Exception exception)
        {
            logger.LogWarning(exception, "Аудит: PushMessage {TenantId} → не принято (сбой схемы/БД)", tenant.Id);
            return new PushMessageReply();
        }
        finally
        {
            tenantContext.Reset();
        }
    }

    /// <summary>
    /// SyncDialogs — синхронизация каталога диалогов аккаунта.
    /// </summary>
    /// <param name="request">Актуальный каталог диалогов (entries).</param>
    /// <param name="context">Контекст вызова.</param>
    /// <returns>monitored_ids — диалоги с включённым мониторингом после применения каталога.</returns>
    public override async Task<SyncDialogsReply> SyncDialogs(SyncDialogsRequest request, ServerCallContext context)
    {
        TenantRecordDto? tenant = await ResolveTenantAsync(context).ConfigureAwait(false);
        if (tenant is null)
        {
            return new SyncDialogsReply();
        }

        await using AsyncServiceScope tenantScope = scopeFactory.CreateAsyncScope();
        ITenantContext tenantContext = tenantScope.ServiceProvider.GetRequiredService<ITenantContext>();
        try
        {
            tenantContext.SetTenant(new TenantId(tenant.Id.ToString("N")));
            DialogsService dialogs = tenantScope.ServiceProvider.GetRequiredService<DialogsService>();

            List<TelegramDialogEntryDto> entries = new(request.Entries.Count);
            foreach (DialogEntry entry in request.Entries)
            {
                entries.Add(new TelegramDialogEntryDto(entry.Id, entry.Name, entry.Username, entry.Kind, entry.Hue));
            }

            int synced = await dialogs.SyncFromTelegramAsync(entries, context.CancellationToken).ConfigureAwait(false);
            IReadOnlyCollection<string> monitoredIds =
                await dialogs.ListMonitoredIdsAsync(context.CancellationToken).ConfigureAwait(false);

            logger.LogInformation(
                "Аудит: SyncDialogs {TenantId}: каталог {Count} → применено {Synced}, monitored {Monitored}",
                tenant.Id,
                request.Entries.Count,
                synced,
                monitoredIds.Count);

            var reply = new SyncDialogsReply();
            reply.MonitoredIds.AddRange(monitoredIds);
            return reply;
        }
        catch (OperationCanceledException)
        {
            throw;
        }
        catch (Exception exception)
        {
            logger.LogWarning(exception, "Аудит: SyncDialogs {TenantId} → каталог не применён (сбой схемы/БД)", tenant.Id);
            return new SyncDialogsReply();
        }
        finally
        {
            tenantContext.Reset();
        }
    }

    /// <summary>
    /// ReportStatus — статус аккаунта в KV + SSE system_status/тосты на переходах фаз.
    /// </summary>
    /// <param name="request">Статус аккаунта из.</param>
    /// <param name="context">Контекст вызова.</param>
    /// <returns>ok — статус принят и сохранён.</returns>
    public override async Task<ReportStatusReply> ReportStatus(ReportStatusRequest request, ServerCallContext context)
    {
        TenantRecordDto? tenant = await ResolveTenantAsync(context).ConfigureAwait(false);
        if (tenant is null)
        {
            return new ReportStatusReply();
        }

        await using AsyncServiceScope tenantScope = scopeFactory.CreateAsyncScope();
        ITenantContext tenantContext = tenantScope.ServiceProvider.GetRequiredService<ITenantContext>();
        try
        {
            tenantContext.SetTenant(new TenantId(tenant.Id.ToString("N")));
            ISettingsStore settings = tenantScope.ServiceProvider.GetRequiredService<ISettingsStore>();

            TgReportedStatus current = ToReportedStatus(request);
            TgReportedStatus? previous = await ReadPreviousStatusAsync(settings, context.CancellationToken).ConfigureAwait(false);

            // SSE до записи KV: канал тенанта обновляется и при сбое записи (следующий репорт перепишет KV).
            PublishStatusEvents(tenant.Id, previous, current);

            await settings.SetAsync(SettingsKeys.TgStatus, ToJson(current), context.CancellationToken).ConfigureAwait(false);
            await settings.SetAsync(SettingsKeys.TgAccount, JsonSerializer.Serialize(request.Account), context.CancellationToken).ConfigureAwait(false);

            logger.LogInformation(
                "Аудит: ReportStatus {TenantId} → фаза {Phase}, connected {Connected}",
                tenant.Id,
                request.Phase,
                request.Connected);
            return new ReportStatusReply { Ok = true };
        }
        catch (OperationCanceledException)
        {
            throw;
        }
        catch (Exception exception)
        {
            logger.LogWarning(exception, "Аудит: ReportStatus {TenantId} → не сохранён (сбой схемы/БД)", tenant.Id);
            return new ReportStatusReply();
        }
        finally
        {
            tenantContext.Reset();
        }
    }

    private async Task<TenantRecordDto?> ResolveTenantAsync(ServerCallContext context)
    {
        string tenantId = RequireTenantIdMetadata(context);
        if (!Guid.TryParse(tenantId, out Guid tenantGuid))
        {
            logger.LogWarning("Аудит: ингресс {Action} → тенант {TenantId} неизвестен (id не Guid)", context.Method, tenantId);
            return null;
        }

        await using AsyncServiceScope registryScope = scopeFactory.CreateAsyncScope();
        ITenantRepository repository = registryScope.ServiceProvider.GetRequiredService<ITenantRepository>();
        TenantRecordDto? tenant = await repository.FindByIdAsync(tenantGuid, context.CancellationToken).ConfigureAwait(false);
        if (tenant is null)
        {
            logger.LogWarning("Аудит: ингресс {Action} → тенант {TenantId} неизвестен (нет в реестре)", context.Method, tenantId);
        }

        return tenant;
    }

    // Читает tenant-id из metadata (обязателен; отсутствие — UNAUTHENTICATED, README).
    // context: Контекст вызова.
    // Возвращает: Значение tenant-id.
    private static string RequireTenantIdMetadata(ServerCallContext context)
    {
        string? tenantId = context.RequestHeaders.GetValue(TenantIdMetadataKey);
        if (string.IsNullOrWhiteSpace(tenantId))
        {
            throw new RpcException(new Status(StatusCode.Unauthenticated, MissingTenantIdDetail));
        }

        return tenantId;
    }

    private async Task SavePreviewSafelyAsync(
        AsyncServiceScope tenantScope,
        TenantRecordDto tenant,
        PushMessageRequest request,
        CancellationToken ct)
    {
        try
        {
            DialogsService dialogs = tenantScope.ServiceProvider.GetRequiredService<DialogsService>();
            DateTimeOffset? msgAt = request.HasMsgAt ? DateTimeOffset.FromUnixTimeMilliseconds(request.MsgAt) : null;
            await dialogs.SavePreviewAsync(
                request.DialogId,
                request.HasMsgId ? request.MsgId : null,
                request.Text,
                msgAt,
                ct).ConfigureAwait(false);
        }
        catch (OperationCanceledException)
        {
            throw;
        }
        catch (Exception exception)
        {
            // Превью — вторичная запись: приём не затронут (лог дебага, не ошибка RPC).
            logger.LogDebug(exception, "PushMessage {TenantId}: превью не сохранено (приём не затронут)", tenant.Id);
        }
    }

    // Публикует SSE system_status (каждый репорт) и тосты на переходах connected.
    // tenantId: Тенант канала (реестровый Guid).
    // previous: Предыдущий снимок из KV (null — первый репорт).
    // current: Текущий снимок репорта.
    private void PublishStatusEvents(
        Guid tenantId,
        TgReportedStatus? previous,
        TgReportedStatus current)
    {
        broker.Publish(tenantId, SystemStatusEventType, current);
        if (previous is null)
        {
            // Первый репорт после старта сервиса: переходов нет, статус фронт получит по system_status.
            return;
        }

        if (!previous.Connected && current.Connected)
        {
            PublishToast(tenantId, ConnectedToastText, ConnectedToastIcon);
        }
        else if (previous.Connected && !current.Connected)
        {
            PublishToast(tenantId, DisconnectedToastText, DisconnectedToastIcon);
        }
    }

    private void PublishToast(
        Guid tenantId,
        string text,
        string icon)
    {
        broker.Publish(tenantId, ToastEventType, new { text, icon });
    }

    // Снимок предыдущего статуса из KV tgStatus (нет записи/битый JSON — null).
    // settings: KV-хранилище настроек схемы тенанта.
    // ct: Токен отмены.
    // Возвращает: Предыдущий снимок либо null.
    private async Task<TgReportedStatus?> ReadPreviousStatusAsync(ISettingsStore settings, CancellationToken ct)
    {
        SettingValue? stored = await settings.GetAsync(SettingsKeys.TgStatus, ct).ConfigureAwait(false);
        if (stored is null)
        {
            return null;
        }

        try
        {
            return JsonSerializer.Deserialize<TgReportedStatus>(stored.ValueJson, StatusJsonOptions);
        }
        catch (JsonException exception)
        {
            logger.LogWarning(exception, "Аудит: KV tgStatus повреждён — переходы фаз не определяются");
            return null;
        }
    }

    // Маппит запрос ReportStatus в снимок KV (account живёт отдельным ключом tgAccount).
    // request: Запрос ReportStatus.
    // Возвращает: Снимок статуса.
    private static TgReportedStatus ToReportedStatus(ReportStatusRequest request) => new()
    {
        Phase = request.Phase,
        Connected = request.Connected,
        Listener = request.Listener,
        Error = request.HasError ? request.Error : null,
        QrUrl = request.HasQrUrl ? request.QrUrl : null,
    };

    // Сериализует снимок в JSON (camelCase, конвенция value_json).
    // status: Снимок статуса.
    // Возвращает: JSON-строка.
    private static string ToJson(TgReportedStatus status) => JsonSerializer.Serialize(status, StatusJsonOptions);
}