sevenhill
343 lines
12 KiB
C#
343 lines
12 KiB
C#
using Microsoft.AspNetCore.SignalR;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using QMax.Api.Data;
|
|
using QMax.Api.Data.Entities;
|
|
using QMax.Api.Infrastructure.Hubs;
|
|
using QMax.Api.Infrastructure.Max;
|
|
using QMax.Api.Infrastructure.Storage;
|
|
|
|
namespace QMax.Api.Services;
|
|
|
|
public sealed class MaxOutboxService(
|
|
QMaxDbContext db,
|
|
IMaxBridgeClient maxBridge,
|
|
IAttachmentStorageService storage,
|
|
ChatProjectionService projection,
|
|
IHubContext<QMaxHub> hubContext,
|
|
ILogger<MaxOutboxService> logger)
|
|
{
|
|
private static readonly TimeSpan InitialChatActionRetryDelay = TimeSpan.FromMinutes(10);
|
|
private static readonly TimeSpan MaxChatActionRetryDelay = TimeSpan.FromHours(6);
|
|
|
|
public async Task<int> ProcessOnceAsync(CancellationToken cancellationToken)
|
|
{
|
|
var processed = await ProcessPendingMessagesOnceAsync(cancellationToken);
|
|
return processed + await ProcessPendingChatActionsOnceAsync(cancellationToken);
|
|
}
|
|
|
|
public async Task<int> ProcessPendingMessagesOnceAsync(CancellationToken cancellationToken)
|
|
{
|
|
var candidates = await db.Messages
|
|
.Include(message => message.Chat)
|
|
.Include(message => message.Attachments)
|
|
.Include(message => message.Reactions)
|
|
.Where(message =>
|
|
message.Direction == MessageDirection.Outgoing &&
|
|
message.DeliveryState == MessageDeliveryState.Sending &&
|
|
message.Chat != null &&
|
|
message.Chat.DeletedAt == null &&
|
|
message.DeletedAt == null)
|
|
.ToListAsync(cancellationToken);
|
|
|
|
var pending = candidates
|
|
.OrderBy(message => message.SentAt)
|
|
.ThenBy(message => message.Id)
|
|
.Take(10)
|
|
.ToArray();
|
|
|
|
var processed = 0;
|
|
foreach (var message in pending)
|
|
{
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
if (await ProcessMessageAsync(message, cancellationToken))
|
|
{
|
|
processed++;
|
|
}
|
|
}
|
|
|
|
return processed;
|
|
}
|
|
|
|
public async Task<int> ProcessPendingChatActionsOnceAsync(CancellationToken cancellationToken)
|
|
{
|
|
var now = DateTimeOffset.UtcNow;
|
|
var pending = await db.Chats
|
|
.Where(chat => chat.PendingMaxAction != null)
|
|
.ToListAsync(cancellationToken);
|
|
pending = pending
|
|
.Where(chat => IsChatActionDue(chat, now))
|
|
.OrderBy(chat => chat.PendingMaxActionRequestedAt ?? chat.UpdatedAt)
|
|
.ThenBy(chat => chat.Id)
|
|
.Take(10)
|
|
.ToList();
|
|
|
|
var processed = 0;
|
|
foreach (var chat in pending)
|
|
{
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
if (await ProcessChatActionAsync(chat, cancellationToken))
|
|
{
|
|
processed++;
|
|
}
|
|
}
|
|
|
|
return processed;
|
|
}
|
|
|
|
private static bool IsChatActionDue(Chat chat, DateTimeOffset now)
|
|
{
|
|
if (chat.PendingMaxActionLastAttemptAt is null)
|
|
{
|
|
return true;
|
|
}
|
|
|
|
return now - chat.PendingMaxActionLastAttemptAt.Value >= GetChatActionRetryDelay(chat.PendingMaxActionAttempts);
|
|
}
|
|
|
|
private static TimeSpan GetChatActionRetryDelay(int attempts)
|
|
{
|
|
if (attempts <= 1)
|
|
{
|
|
return InitialChatActionRetryDelay;
|
|
}
|
|
|
|
var multiplier = Math.Pow(2, Math.Min(attempts - 1, 5));
|
|
var retryDelay = TimeSpan.FromTicks((long)(InitialChatActionRetryDelay.Ticks * multiplier));
|
|
return retryDelay <= MaxChatActionRetryDelay
|
|
? retryDelay
|
|
: MaxChatActionRetryDelay;
|
|
}
|
|
|
|
private async Task<bool> ProcessChatActionAsync(Chat chat, CancellationToken cancellationToken)
|
|
{
|
|
var pendingAction = chat.PendingMaxAction;
|
|
if (pendingAction is null)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
MaxActionResult result;
|
|
try
|
|
{
|
|
result = pendingAction.Value switch
|
|
{
|
|
ChatPendingMaxAction.ClearHistory => await ApplyMaxChatActionAsync(
|
|
chat,
|
|
(externalChatId, chatUrl) => maxBridge.ClearChatHistoryAsync(externalChatId, chatUrl, cancellationToken),
|
|
cancellationToken),
|
|
ChatPendingMaxAction.DeleteChat => new MaxActionResult(
|
|
false,
|
|
"Chat deletion is disabled because MAX does not remove chats from the web client."),
|
|
_ => new MaxActionResult(false, $"Unsupported chat action: {pendingAction.Value}.")
|
|
};
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
logger.LogWarning(ex, "MAX chat action {Action} failed for chat {ChatId}.", pendingAction.Value, chat.Id);
|
|
result = new MaxActionResult(false, ex.Message);
|
|
}
|
|
|
|
var now = DateTimeOffset.UtcNow;
|
|
chat.PendingMaxActionLastAttemptAt = now;
|
|
if (result.Success)
|
|
{
|
|
chat.PendingMaxAction = null;
|
|
chat.PendingMaxActionRequestedAt = null;
|
|
chat.PendingMaxActionLastAttemptAt = null;
|
|
chat.PendingMaxActionAttempts = 0;
|
|
chat.PendingMaxActionError = null;
|
|
}
|
|
else
|
|
{
|
|
chat.PendingMaxActionAttempts++;
|
|
chat.PendingMaxActionError = result.Error ?? $"MAX chat action {pendingAction.Value} failed.";
|
|
}
|
|
|
|
chat.UpdatedAt = now;
|
|
await db.SaveChangesAsync(cancellationToken);
|
|
if (!result.Success)
|
|
{
|
|
logger.LogWarning(
|
|
"MAX chat action {Action} failed for chat {ChatId}: {Error}",
|
|
pendingAction.Value,
|
|
chat.Id,
|
|
chat.PendingMaxActionError);
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
private async Task<MaxActionResult> ApplyMaxChatActionAsync(
|
|
Chat chat,
|
|
Func<string, string?, Task<MaxActionResult>> action,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(chat.ExternalId))
|
|
{
|
|
return new MaxActionResult(true, null);
|
|
}
|
|
|
|
var chatUrl = chat.WebUrl;
|
|
if (string.IsNullOrWhiteSpace(chatUrl))
|
|
{
|
|
chatUrl = await ResolveAndPersistChatUrlAsync(chat, cancellationToken);
|
|
}
|
|
|
|
var result = await action(chat.ExternalId, chatUrl);
|
|
if (!result.Success && IsChatOpenFailure(result.Error))
|
|
{
|
|
var previousUrl = chatUrl;
|
|
chatUrl = await ResolveAndPersistChatUrlAsync(chat, cancellationToken);
|
|
if (!string.IsNullOrWhiteSpace(chatUrl) &&
|
|
!string.Equals(previousUrl, chatUrl, StringComparison.Ordinal))
|
|
{
|
|
result = await action(chat.ExternalId, chatUrl);
|
|
}
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
private async Task<string?> ResolveAndPersistChatUrlAsync(Chat chat, CancellationToken cancellationToken)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(chat.ExternalId))
|
|
{
|
|
return chat.WebUrl;
|
|
}
|
|
|
|
var result = await maxBridge.ResolveChatUrlAsync(chat.ExternalId, cancellationToken);
|
|
if (!result.Success || string.IsNullOrWhiteSpace(result.ChatUrl))
|
|
{
|
|
return chat.WebUrl;
|
|
}
|
|
|
|
var nextWebUrl = result.ChatUrl.Trim();
|
|
if (!string.Equals(chat.WebUrl, nextWebUrl, StringComparison.Ordinal))
|
|
{
|
|
chat.WebUrl = nextWebUrl;
|
|
}
|
|
|
|
return chat.WebUrl;
|
|
}
|
|
|
|
private static bool IsChatOpenFailure(string? error)
|
|
{
|
|
return error?.Contains("not found or did not open", StringComparison.OrdinalIgnoreCase) == true ||
|
|
error?.Contains("menu was not opened", StringComparison.OrdinalIgnoreCase) == true;
|
|
}
|
|
|
|
private async Task<bool> ProcessMessageAsync(Message message, CancellationToken cancellationToken)
|
|
{
|
|
var chat = message.Chat;
|
|
if (chat is null || string.IsNullOrWhiteSpace(chat.ExternalId))
|
|
{
|
|
await MarkFailedAsync(message, "Chat is not linked to MAX.", cancellationToken);
|
|
return true;
|
|
}
|
|
|
|
MaxSendResult result;
|
|
try
|
|
{
|
|
result = await SendToMaxAsync(chat, message, cancellationToken);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
logger.LogWarning(ex, "MAX outbox send failed for message {MessageId}.", message.Id);
|
|
result = new MaxSendResult(false, null, ex.Message);
|
|
}
|
|
|
|
if (result.Success)
|
|
{
|
|
message.ExternalId = string.IsNullOrWhiteSpace(result.ExternalMessageId)
|
|
? message.ExternalId
|
|
: result.ExternalMessageId;
|
|
message.DeliveryState = MessageDeliveryState.Sent;
|
|
message.Error = null;
|
|
if (!string.IsNullOrWhiteSpace(result.ChatUrl) &&
|
|
!string.Equals(chat.WebUrl, result.ChatUrl, StringComparison.Ordinal))
|
|
{
|
|
chat.WebUrl = result.ChatUrl;
|
|
chat.UpdatedAt = DateTimeOffset.UtcNow;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
message.DeliveryState = MessageDeliveryState.Failed;
|
|
message.Error = result.Error ?? "MAX did not accept the outgoing message.";
|
|
}
|
|
|
|
await db.SaveChangesAsync(cancellationToken);
|
|
await PublishMessageUpdatedAsync(message, cancellationToken);
|
|
return true;
|
|
}
|
|
|
|
private async Task<MaxSendResult> SendToMaxAsync(Chat chat, Message message, CancellationToken cancellationToken)
|
|
{
|
|
var externalChatId = chat.ExternalId!;
|
|
var chatUrl = chat.WebUrl;
|
|
var attachments = message.Attachments.OrderBy(attachment => attachment.SortOrder).ToArray();
|
|
if (attachments.Length == 0)
|
|
{
|
|
return string.IsNullOrWhiteSpace(message.Text)
|
|
? new MaxSendResult(false, null, "Text is empty.")
|
|
: await maxBridge.SendTextAsync(externalChatId, chatUrl, message.Text, cancellationToken);
|
|
}
|
|
|
|
MaxSendResult? lastResult = null;
|
|
var errors = new List<string>();
|
|
foreach (var attachment in attachments)
|
|
{
|
|
var path = storage.GetPath(attachment.StorageFileName);
|
|
if (!File.Exists(path))
|
|
{
|
|
errors.Add($"Attachment file is missing: {attachment.OriginalFileName}");
|
|
continue;
|
|
}
|
|
|
|
var caption = attachment.SortOrder == 0 ? message.Text ?? "" : "";
|
|
var result = await maxBridge.SendAttachmentAsync(externalChatId, chatUrl, path, caption, cancellationToken);
|
|
if (!string.IsNullOrWhiteSpace(result.ChatUrl))
|
|
{
|
|
chatUrl = result.ChatUrl;
|
|
}
|
|
lastResult = result;
|
|
if (!result.Success && !string.IsNullOrWhiteSpace(result.Error))
|
|
{
|
|
errors.Add(result.Error);
|
|
}
|
|
}
|
|
|
|
return errors.Count == 0
|
|
? new MaxSendResult(true, lastResult?.ExternalMessageId, null, chatUrl)
|
|
: new MaxSendResult(false, lastResult?.ExternalMessageId, string.Join("; ", errors.Distinct()), chatUrl);
|
|
}
|
|
|
|
private async Task MarkFailedAsync(Message message, string error, CancellationToken cancellationToken)
|
|
{
|
|
message.DeliveryState = MessageDeliveryState.Failed;
|
|
message.Error = error;
|
|
await db.SaveChangesAsync(cancellationToken);
|
|
await PublishMessageUpdatedAsync(message, cancellationToken);
|
|
}
|
|
|
|
private async Task PublishMessageUpdatedAsync(Message message, CancellationToken cancellationToken)
|
|
{
|
|
var updated = await db.Messages
|
|
.AsNoTracking()
|
|
.Include(x => x.Attachments)
|
|
.Include(x => x.Reactions)
|
|
.FirstOrDefaultAsync(x => x.Id == message.Id, cancellationToken);
|
|
|
|
if (updated is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
await hubContext.Clients.Group($"chat:{updated.ChatId}").SendAsync(
|
|
"MessageUpdated",
|
|
projection.ToDto(updated),
|
|
cancellationToken);
|
|
await hubContext.Clients.All.SendAsync("ChatListInvalidated", cancellationToken);
|
|
}
|
|
}
|