stepan
b8570e3197
ci / build-test (pull_request) Successful in 2m53s
Описание живёт только в интерфейсе: удалены классовые summary и док-блоки членов-реализаций (28 файлов). Правило уточнено в §11.
453 lines
17 KiB
C#
453 lines
17 KiB
C#
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;
|
||
|
||
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;
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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");
|
||
}
|
||
}
|
||
|
||
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();
|
||
}
|
||
}
|