Add SQL outbox for MAX sends
This commit is contained in:
sevenhill
@@ -210,6 +210,37 @@ public sealed class MaxBridgeSyncService(
|
||||
continue;
|
||||
}
|
||||
|
||||
var pendingOutgoing = await FindPendingOutgoingEchoAsync(db, chat.Id, incoming, cancellationToken);
|
||||
if (pendingOutgoing is not null)
|
||||
{
|
||||
var merge = await MergeExistingMessageAsync(
|
||||
db,
|
||||
pendingOutgoing,
|
||||
incoming,
|
||||
maxBridgeClient,
|
||||
storage,
|
||||
cancellationToken);
|
||||
if (!string.Equals(pendingOutgoing.ExternalId, incoming.ExternalId, StringComparison.Ordinal))
|
||||
{
|
||||
pendingOutgoing.ExternalId = incoming.ExternalId;
|
||||
merge = merge with { Changed = true };
|
||||
}
|
||||
|
||||
if (merge.Changed)
|
||||
{
|
||||
updatedMessages.Add((chat, pendingOutgoing));
|
||||
if (chat.LastMessageAt is null || pendingOutgoing.SentAt >= chat.LastMessageAt)
|
||||
{
|
||||
chat.LastMessagePreview = LastMessagePreview(pendingOutgoing.Text, pendingOutgoing.Attachments.Concat(merge.AddedAttachments));
|
||||
chat.LastMessageAt = pendingOutgoing.SentAt;
|
||||
}
|
||||
|
||||
changed++;
|
||||
}
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
var message = new Message
|
||||
{
|
||||
Chat = chat,
|
||||
@@ -528,6 +559,89 @@ public sealed class MaxBridgeSyncService(
|
||||
|
||||
private sealed record MessageMergeResult(bool Changed, IReadOnlyList<MessageAttachment> AddedAttachments);
|
||||
|
||||
private static async Task<Message?> FindPendingOutgoingEchoAsync(
|
||||
QMaxDbContext db,
|
||||
Guid chatId,
|
||||
MaxMessageUpdate incoming,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (!incoming.IsOutgoing || string.IsNullOrWhiteSpace(incoming.ExternalId))
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
var candidates = await db.Messages
|
||||
.Include(x => x.Attachments)
|
||||
.Where(x =>
|
||||
x.ChatId == chatId &&
|
||||
x.DeletedAt == null &&
|
||||
x.Direction == MessageDirection.Outgoing &&
|
||||
(x.DeliveryState == MessageDeliveryState.Sending ||
|
||||
x.DeliveryState == MessageDeliveryState.Failed) &&
|
||||
(x.ExternalId == null || x.ExternalId == ""))
|
||||
.ToListAsync(cancellationToken);
|
||||
if (candidates.Count == 0)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
var incomingAttachments = DistinctAttachments(incoming.Attachments ?? []);
|
||||
var incomingText = MessageTextSanitizer.CleanMessageText(
|
||||
incoming.Text,
|
||||
incomingAttachments.Count > 0);
|
||||
|
||||
return candidates
|
||||
.Where(candidate => IsRecentOutgoingEchoCandidate(candidate, incoming))
|
||||
.Where(candidate => IsPendingOutgoingEchoMatch(candidate, incomingAttachments, incomingText))
|
||||
.OrderBy(candidate => Math.Abs((candidate.SentAt.ToUniversalTime() - incoming.SentAt.ToUniversalTime()).TotalSeconds))
|
||||
.FirstOrDefault();
|
||||
}
|
||||
|
||||
private static bool IsRecentOutgoingEchoCandidate(Message candidate, MaxMessageUpdate incoming)
|
||||
{
|
||||
var candidateAt = candidate.SentAt.ToUniversalTime();
|
||||
var incomingAt = incoming.SentAt.ToUniversalTime();
|
||||
return Math.Abs((candidateAt - incomingAt).TotalMinutes) <= 30 ||
|
||||
candidateAt >= DateTimeOffset.UtcNow.AddMinutes(-30);
|
||||
}
|
||||
|
||||
private static bool IsPendingOutgoingEchoMatch(
|
||||
Message candidate,
|
||||
IReadOnlyList<MaxAttachmentUpdate> incomingAttachments,
|
||||
string? incomingText)
|
||||
{
|
||||
var localAttachments = candidate.Attachments.OrderBy(x => x.SortOrder).ToArray();
|
||||
var localText = MessageTextSanitizer.CleanMessageText(candidate.Text, localAttachments.Length > 0);
|
||||
|
||||
if (incomingAttachments.Count == 0)
|
||||
{
|
||||
return localAttachments.Length == 0 &&
|
||||
string.Equals(localText, incomingText, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
if (localAttachments.Length == 0)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(localText) &&
|
||||
!string.IsNullOrWhiteSpace(incomingText) &&
|
||||
!string.Equals(localText, incomingText, StringComparison.Ordinal))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
return incomingAttachments.All(incomingAttachment =>
|
||||
{
|
||||
var kind = ResolveKind(incomingAttachment) ?? AttachmentKind.File;
|
||||
return localAttachments.Any(localAttachment =>
|
||||
localAttachment.Kind == kind &&
|
||||
(localAttachment.SortOrder == incomingAttachment.SortOrder ||
|
||||
localAttachments.Length == 1 ||
|
||||
incomingAttachments.Count == 1));
|
||||
});
|
||||
}
|
||||
|
||||
private static async Task<MessageMergeResult> MergeExistingMessageAsync(
|
||||
QMaxDbContext db,
|
||||
Message message,
|
||||
|
||||
@@ -0,0 +1,151 @@
|
||||
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)
|
||||
{
|
||||
public async Task<int> ProcessOnceAsync(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.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;
|
||||
}
|
||||
|
||||
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.ExternalId, 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;
|
||||
}
|
||||
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(string externalChatId, Message message, CancellationToken cancellationToken)
|
||||
{
|
||||
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, 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, path, caption, cancellationToken);
|
||||
lastResult = result;
|
||||
if (!result.Success && !string.IsNullOrWhiteSpace(result.Error))
|
||||
{
|
||||
errors.Add(result.Error);
|
||||
}
|
||||
}
|
||||
|
||||
return errors.Count == 0
|
||||
? new MaxSendResult(true, lastResult?.ExternalMessageId, null)
|
||||
: new MaxSendResult(false, lastResult?.ExternalMessageId, string.Join("; ", errors.Distinct()));
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
using Microsoft.Extensions.Options;
|
||||
using QMax.Api.Configuration;
|
||||
|
||||
namespace QMax.Api.Services;
|
||||
|
||||
public sealed class MaxOutboxWorker(
|
||||
IServiceScopeFactory scopeFactory,
|
||||
IOptions<QMaxOptions> options,
|
||||
ILogger<MaxOutboxWorker> logger) : BackgroundService
|
||||
{
|
||||
private readonly QMaxOptions _options = options.Value;
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
var delay = TimeSpan.FromSeconds(Math.Max(1, _options.MaxOutboxPollIntervalSeconds));
|
||||
using var timer = new PeriodicTimer(delay);
|
||||
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
await timer.WaitForNextTickAsync(stoppingToken);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await using var scope = scopeFactory.CreateAsyncScope();
|
||||
var outbox = scope.ServiceProvider.GetRequiredService<MaxOutboxService>();
|
||||
await outbox.ProcessOnceAsync(stoppingToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogWarning(ex, "QMAX outbox tick failed.");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user