SaaS-мониторинг Telegram: ядро (модули Cards/Kanban/Pipeline/Tenants/Settings/ Discovery, Api, Infrastructure), сервисы telegram/ai/ml/storage, фронт Vue, контракты и grpc-hosting, деплой-конфиги (dev/prod/observability/CI-раннер), Gitea Actions CI, документация (ТЗ, техдок, api-map, код-стайл, планы, бэклог). Текущее состояние: все этапы роадмапа 0–12 закрыты, сборка 5 sln 0/0, тесты 1340/130/52/38/9 зелёные.
This commit is contained in:
Rustam Khalimov
@@ -0,0 +1,472 @@
|
||||
using Deal.Contracts.Integrations.Abstractions;
|
||||
using Deal.Contracts.Integrations.Models;
|
||||
using Deal.Grpc.Telegram;
|
||||
using Deal.Infrastructure.Integrations.Models;
|
||||
using Deal.SharedKernel.Tenants.Abstractions;
|
||||
using Deal.SharedKernel.Tenants.Models;
|
||||
using Grpc.Core;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Deal.Infrastructure.Integrations.Services;
|
||||
|
||||
/// <summary>
|
||||
/// gRPC-адаптер порта <see cref="ITelegramGateway"/> к автономному telegram-service.
|
||||
/// </summary>
|
||||
public sealed class GrpcTelegramClient : ITelegramGateway
|
||||
{
|
||||
/// <summary>
|
||||
/// Deadline локальных команд статуса/зеркала — 10 с.
|
||||
/// </summary>
|
||||
public const int ShortDeadlineSeconds = 10;
|
||||
|
||||
/// <summary>
|
||||
/// Deadline сетевых команд Telegram — 60 с.
|
||||
/// </summary>
|
||||
public const int CommandDeadlineSeconds = 60;
|
||||
|
||||
/// <summary>
|
||||
/// Deadline тяжёлых команд каталога/backfill — 120 с.
|
||||
/// </summary>
|
||||
public const int LongDeadlineSeconds = 120;
|
||||
|
||||
private const string NotConnectedDetail = "Telegram не подключён";
|
||||
|
||||
private readonly ITenantContext _tenantContext;
|
||||
|
||||
// Транспорт gRPC telegram-service (канал + metadata).
|
||||
private readonly TelegramGrpcConnection _connection;
|
||||
|
||||
// Логгер сбоев вызовов telegram-service.
|
||||
private readonly ILogger<GrpcTelegramClient> _logger;
|
||||
|
||||
/// <summary>
|
||||
/// Создаёт gRPC-адаптер гейта telegram-service.
|
||||
/// </summary>
|
||||
/// <param name="tenantContext">Контекст текущего тенанта (AsyncLocal, заполняется middleware/циклами).</param>
|
||||
/// <param name="connection">Транспорт telegram-service (singleton-канал + service-token).</param>
|
||||
/// <param name="logger">Логгер сбоев.</param>
|
||||
public GrpcTelegramClient(
|
||||
ITenantContext tenantContext,
|
||||
TelegramGrpcConnection connection,
|
||||
ILogger<GrpcTelegramClient> logger)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(tenantContext);
|
||||
ArgumentNullException.ThrowIfNull(connection);
|
||||
ArgumentNullException.ThrowIfNull(logger);
|
||||
_tenantContext = tenantContext;
|
||||
_connection = connection;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task<TelegramAccountStatusDto> ITelegramGateway.StatusAsync(CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
GetStatusReply reply = await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(ShortDeadlineSeconds), ct,
|
||||
(client, options) => client.GetStatusAsync(new GetStatusRequest(), options));
|
||||
return new TelegramAccountStatusDto(
|
||||
Phase: reply.Phase,
|
||||
Connected: reply.Connected,
|
||||
Listener: reply.Listener,
|
||||
Account: reply.Account,
|
||||
Error: reply.HasError ? reply.Error : null,
|
||||
QrUrl: reply.HasQrUrl ? reply.QrUrl : null);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "status");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task<TelegramAuthResultDto> ITelegramGateway.StartPhoneAsync(
|
||||
string phone,
|
||||
int apiId,
|
||||
string apiHash,
|
||||
CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
StartPhoneReply reply = await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(CommandDeadlineSeconds), ct,
|
||||
(client, options) => client.StartPhoneAsync(
|
||||
new StartPhoneRequest { Phone = phone ?? string.Empty, ApiId = apiId, ApiHash = apiHash ?? string.Empty },
|
||||
options));
|
||||
return new TelegramAuthResultDto(Phase: reply.Phase, QrUrl: null);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "start_phone");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task<TelegramAuthResultDto> ITelegramGateway.StartQrAsync(
|
||||
int apiId,
|
||||
string apiHash,
|
||||
CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
StartQrReply reply = await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(CommandDeadlineSeconds), ct,
|
||||
(client, options) => client.StartQrAsync(
|
||||
new StartQrRequest { ApiId = apiId, ApiHash = apiHash ?? string.Empty },
|
||||
options));
|
||||
return new TelegramAuthResultDto(
|
||||
Phase: reply.Phase,
|
||||
QrUrl: string.IsNullOrEmpty(reply.QrUrl) ? null : reply.QrUrl);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "start_qr");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task<string> ITelegramGateway.SendCodeAsync(string code, CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
SendCodeReply reply = await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(CommandDeadlineSeconds), ct,
|
||||
(client, options) => client.SendCodeAsync(
|
||||
new SendCodeRequest { Code = code ?? string.Empty }, options));
|
||||
return reply.Phase;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "send_code");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task<string> ITelegramGateway.SendPasswordAsync(string password, CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
SendPasswordReply reply = await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(CommandDeadlineSeconds), ct,
|
||||
(client, options) => client.SendPasswordAsync(
|
||||
new SendPasswordRequest { Password = password ?? string.Empty }, options));
|
||||
return reply.Phase;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "send_password");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task ITelegramGateway.LogoutAsync(CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(ShortDeadlineSeconds), ct,
|
||||
(client, options) => client.LogoutAsync(new LogoutRequest(), options));
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "logout");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task<IReadOnlyList<TelegramDialogEntryDto>> ITelegramGateway.RefreshDialogsAsync(CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
RefreshDialogsReply reply = await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(LongDeadlineSeconds), ct,
|
||||
(client, options) => client.RefreshDialogsAsync(new RefreshDialogsRequest(), options));
|
||||
return MapEntries(reply.Entries);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "refresh_dialogs");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task ITelegramGateway.SetMonitorAsync(
|
||||
string dialogId,
|
||||
bool enabled,
|
||||
CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(ShortDeadlineSeconds), ct,
|
||||
(client, options) => client.SetMonitorAsync(
|
||||
new SetMonitorRequest { DialogId = dialogId ?? string.Empty, Enabled = enabled }, options));
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "set_monitor");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task ITelegramGateway.SetMonitorAllAsync(bool enabled, CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(ShortDeadlineSeconds), ct,
|
||||
(client, options) => client.SetMonitorAllAsync(
|
||||
new SetMonitorAllRequest { Enabled = enabled }, options));
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "set_monitor_all");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task<int> ITelegramGateway.BackfillAsync(
|
||||
string dialogId,
|
||||
bool force,
|
||||
CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
BackfillReply reply = await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(LongDeadlineSeconds), ct,
|
||||
(client, options) => client.BackfillAsync(
|
||||
new BackfillRequest { DialogId = dialogId ?? string.Empty, Force = force }, options));
|
||||
return reply.Processed;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "backfill");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task<IReadOnlyList<TelegramRecentMessageDto>> ITelegramGateway.ReadRecentAsync(
|
||||
string dialogId,
|
||||
int limit,
|
||||
CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
ReadRecentReply reply = await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(CommandDeadlineSeconds), ct,
|
||||
(client, options) => client.ReadRecentAsync(
|
||||
new ReadRecentRequest { DialogId = dialogId ?? string.Empty, Limit = limit }, options));
|
||||
return reply.Messages
|
||||
.Select(message => new TelegramRecentMessageDto(message.Id, message.Text, message.Time))
|
||||
.ToList();
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "read_recent");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task<TelegramSourceContentDto> ITelegramGateway.ReadSourceAsync(
|
||||
string dialogId,
|
||||
long msgId,
|
||||
CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
ReadSourceReply reply = await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(CommandDeadlineSeconds), ct,
|
||||
(client, options) => client.ReadSourceAsync(
|
||||
new ReadSourceRequest { DialogId = dialogId ?? string.Empty, MsgId = msgId }, options));
|
||||
return new TelegramSourceContentDto(
|
||||
reply.Found,
|
||||
reply.HasText ? reply.Text : null,
|
||||
reply.HasTime ? reply.Time : null);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "read_source");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task<IReadOnlyList<TelegramDialogEntryDto>> ITelegramGateway.SearchAsync(
|
||||
string query,
|
||||
int limit,
|
||||
CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
SearchReply reply = await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(CommandDeadlineSeconds), ct,
|
||||
(client, options) => client.SearchAsync(
|
||||
new SearchRequest { Query = query ?? string.Empty, Limit = limit }, options));
|
||||
return MapEntries(reply.Results);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "search");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task<TelegramChannelInfoDto> ITelegramGateway.InfoAsync(string dialogId, CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
GetInfoReply reply = await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(CommandDeadlineSeconds), ct,
|
||||
(client, options) => client.GetInfoAsync(
|
||||
new GetInfoRequest { DialogId = dialogId ?? string.Empty }, options));
|
||||
ChannelInfo info = reply.Info;
|
||||
return new TelegramChannelInfoDto(
|
||||
Id: info.Id,
|
||||
Name: info.Name,
|
||||
Username: info.Username,
|
||||
Kind: info.Kind,
|
||||
Hue: info.Hue,
|
||||
Participants: info.HasParticipants ? info.Participants : null,
|
||||
IsForum: info.IsForum);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "get_info");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task<TelegramEvalReadDto> ITelegramGateway.ReadForEvalAsync(
|
||||
string dialogId,
|
||||
int limit,
|
||||
CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
ReadForEvalReply reply = await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(CommandDeadlineSeconds), ct,
|
||||
(client, options) => client.ReadForEvalAsync(
|
||||
new ReadForEvalRequest { DialogId = dialogId ?? string.Empty, Limit = limit }, options));
|
||||
return new TelegramEvalReadDto(
|
||||
Ok: reply.Ok,
|
||||
Error: reply.HasError ? reply.Error : null,
|
||||
Messages: reply.Messages
|
||||
.Select(message => new TelegramEvalMessageDto(
|
||||
Id: message.Id,
|
||||
Text: message.Text,
|
||||
DateMs: message.DateMs,
|
||||
TopicId: message.HasTopicId ? message.TopicId : null,
|
||||
TopicTitle: message.HasTopicTitle ? message.TopicTitle : null))
|
||||
.ToList());
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "read_for_eval");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task ITelegramGateway.JoinAsync(string username, CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(CommandDeadlineSeconds), ct,
|
||||
(client, options) => client.JoinAsync(
|
||||
new JoinRequest { Username = username ?? string.Empty }, options));
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "join");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
async Task ITelegramGateway.LeaveAsync(string dialogId, CancellationToken ct)
|
||||
{
|
||||
TenantId tenantId = RequireTenant();
|
||||
try
|
||||
{
|
||||
await CallAsync(
|
||||
tenantId, TimeSpan.FromSeconds(CommandDeadlineSeconds), ct,
|
||||
(client, options) => client.LeaveAsync(
|
||||
new LeaveRequest { DialogId = dialogId ?? string.Empty }, options));
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
throw TranslateTransportFailure(exception, tenantId, "leave");
|
||||
}
|
||||
}
|
||||
|
||||
// Текущий тенант scope (без него metadata вызовов не имеет смысла).
|
||||
// Возвращает: Идентификатор тенанта.
|
||||
// Исключение InvalidOperationException: Вызов вне tenant-контекста (конфигурация).
|
||||
private TenantId RequireTenant()
|
||||
=> _tenantContext.TenantId
|
||||
?? throw new InvalidOperationException(
|
||||
"GrpcTelegramClient запрошен вне tenant-контекста (ITenantContext.TenantId == null).");
|
||||
|
||||
private async Task<TReply> CallAsync<TReply>(
|
||||
TenantId tenantId,
|
||||
TimeSpan deadline,
|
||||
CancellationToken ct,
|
||||
Func<TelegramService.TelegramServiceClient, CallOptions, AsyncUnaryCall<TReply>> call)
|
||||
where TReply : class
|
||||
{
|
||||
TelegramService.TelegramServiceClient client = _connection.CreateClient();
|
||||
var options = new CallOptions(
|
||||
headers: _connection.CreateMetadata(tenantId.Value),
|
||||
deadline: DateTime.UtcNow.Add(deadline),
|
||||
cancellationToken: ct);
|
||||
return await call(client, options);
|
||||
}
|
||||
|
||||
private Exception TranslateTransportFailure(
|
||||
Exception exception,
|
||||
TenantId tenantId,
|
||||
string operation)
|
||||
{
|
||||
// Отмена по токену вызывающего — не ошибка сервиса (пробрасываем как обычно).
|
||||
if (exception is OperationCanceledException)
|
||||
{
|
||||
return exception;
|
||||
}
|
||||
|
||||
if (exception is RpcException rpc &&
|
||||
(rpc.StatusCode != StatusCode.Unavailable || !string.IsNullOrEmpty(rpc.Status.Detail)))
|
||||
{
|
||||
return rpc;
|
||||
}
|
||||
|
||||
_logger.LogWarning(exception, "Telegram {Operation} недоступен (тенант {TenantId})", operation, tenantId.Value);
|
||||
return new RpcException(new Status(StatusCode.Unavailable, NotConnectedDetail));
|
||||
}
|
||||
|
||||
private static IReadOnlyList<TelegramDialogEntryDto> MapEntries(Google.Protobuf.Collections.RepeatedField<DialogEntry> entries)
|
||||
{
|
||||
return entries
|
||||
.Select(entry => new TelegramDialogEntryDto(
|
||||
Id: entry.Id,
|
||||
Name: entry.Name,
|
||||
Handle: entry.Username,
|
||||
Kind: entry.Kind,
|
||||
Hue: entry.Hue))
|
||||
.ToList();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user