diff --git a/server/QMax.Api/Configuration/QMaxOptions.cs b/server/QMax.Api/Configuration/QMaxOptions.cs index fdf21cf..3884137 100644 --- a/server/QMax.Api/Configuration/QMaxOptions.cs +++ b/server/QMax.Api/Configuration/QMaxOptions.cs @@ -15,6 +15,7 @@ public sealed class QMaxOptions public string MaxUserDataPath { get; set; } = "data/max-profile"; public bool MaxHeadless { get; set; } = true; public int MaxPollIntervalSeconds { get; set; } = 6; + public int MaxOutboxPollIntervalSeconds { get; set; } = 1; public long MaxUploadBytes { get; set; } = 25L * 1024 * 1024; public string CorsAllowedOrigins { get; set; } = ""; public string ReleasesPath { get; set; } = "data/releases"; diff --git a/server/QMax.Api/Controllers/ChatsController.cs b/server/QMax.Api/Controllers/ChatsController.cs index 3ecd55b..1291861 100644 --- a/server/QMax.Api/Controllers/ChatsController.cs +++ b/server/QMax.Api/Controllers/ChatsController.cs @@ -223,21 +223,14 @@ public sealed class ChatsController( return BadRequest("Chat is not linked to MAX."); } - var result = await maxBridge.SendTextAsync(chat.ExternalId, text, cancellationToken); - if (!result.Success) - { - return MaxSendFailure(result, "text message"); - } - var message = new Message { ChatId = chat.Id, - ExternalId = result.ExternalMessageId, Direction = MessageDirection.Outgoing, Text = text, ReplyToMessageId = request.ReplyToMessageId, SentAt = DateTimeOffset.UtcNow, - DeliveryState = MessageDeliveryState.Sent, + DeliveryState = MessageDeliveryState.Sending, SenderName = "You" }; db.Messages.Add(message); @@ -507,14 +500,6 @@ public sealed class ChatsController( }); } - var result = await SendForwardedToMaxAsync(target.ExternalId, message, cancellationToken); - if (!result.Success) - { - return MaxSendFailure(result, "forward"); - } - - message.ExternalId = result.ExternalMessageId; - message.DeliveryState = MessageDeliveryState.Sent; db.Messages.Add(message); target.LastMessagePreview = ForwardPreview(message); target.LastMessageAt = message.SentAt; @@ -548,21 +533,15 @@ public sealed class ChatsController( } var stored = await storage.SaveAsync(file, cancellationToken); - var result = await maxBridge.SendAttachmentAsync(chat.ExternalId, storage.GetPath(stored.StorageFileName), caption ?? "", cancellationToken); - if (!result.Success) - { - return MaxSendFailure(result, "attachment"); - } var message = new Message { ChatId = chat.Id, - ExternalId = result.ExternalMessageId, Direction = MessageDirection.Outgoing, Text = caption, ReplyToMessageId = replyToMessageId, SentAt = DateTimeOffset.UtcNow, - DeliveryState = MessageDeliveryState.Sent, + DeliveryState = MessageDeliveryState.Sending, SenderName = "You", Attachments = [ @@ -652,13 +631,6 @@ public sealed class ChatsController( return PhysicalFile(path, attachment.ContentType, attachment.OriginalFileName, enableRangeProcessing: true); } - private ObjectResult MaxSendFailure(MaxSendResult result, string itemName) - { - return Problem( - result.Error ?? $"MAX did not accept the {itemName}.", - statusCode: StatusCodes.Status502BadGateway); - } - private async Task TryApplyMaxActionAsync( Message message, Func> action, @@ -750,34 +722,6 @@ public sealed class ChatsController( throw new InvalidOperationException($"Cannot forward attachment {attachment.Id}: content is unavailable."); } - private async Task SendForwardedToMaxAsync(string externalChatId, Message message, CancellationToken cancellationToken) - { - var attachments = message.Attachments.OrderBy(x => x.SortOrder).ToArray(); - if (attachments.Length == 0) - { - return string.IsNullOrWhiteSpace(message.Text) - ? new MaxSendResult(false, null, "Forwarded message is empty.") - : await maxBridge.SendTextAsync(externalChatId, message.Text, cancellationToken); - } - - MaxSendResult? lastResult = null; - var errors = new List(); - foreach (var attachment in attachments) - { - var caption = attachment.SortOrder == 0 ? message.Text ?? "" : ""; - var result = await maxBridge.SendAttachmentAsync(externalChatId, storage.GetPath(attachment.StorageFileName), 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 static string? ResolveForwardedFrom(Message source) { return source.SenderName?.Trim().Length > 0 diff --git a/server/QMax.Api/Program.cs b/server/QMax.Api/Program.cs index 276548a..b30b5c2 100644 --- a/server/QMax.Api/Program.cs +++ b/server/QMax.Api/Program.cs @@ -95,7 +95,9 @@ builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); +builder.Services.AddScoped(); builder.Services.AddHostedService(); +builder.Services.AddHostedService(); if (string.Equals(configuredOptions.MaxMode, "Mock", StringComparison.OrdinalIgnoreCase)) { @@ -176,6 +178,11 @@ static async Task EnsureCompatibilitySchemaAsync(QMaxDbContext db) ON Messages (ChatId, ExternalId); """); + await db.Database.ExecuteSqlRawAsync(""" + CREATE INDEX IF NOT EXISTS IX_Messages_DeliveryState_SentAt + ON Messages (DeliveryState, SentAt); + """); + await db.Database.ExecuteSqlRawAsync(""" CREATE TABLE IF NOT EXISTS MessageReactions ( Id TEXT NOT NULL CONSTRAINT PK_MessageReactions PRIMARY KEY, diff --git a/server/QMax.Api/Services/MaxBridgeSyncService.cs b/server/QMax.Api/Services/MaxBridgeSyncService.cs index fbd76d4..f2a2404 100644 --- a/server/QMax.Api/Services/MaxBridgeSyncService.cs +++ b/server/QMax.Api/Services/MaxBridgeSyncService.cs @@ -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 AddedAttachments); + private static async Task 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 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 MergeExistingMessageAsync( QMaxDbContext db, Message message, diff --git a/server/QMax.Api/Services/MaxOutboxService.cs b/server/QMax.Api/Services/MaxOutboxService.cs new file mode 100644 index 0000000..1225b4a --- /dev/null +++ b/server/QMax.Api/Services/MaxOutboxService.cs @@ -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 hubContext, + ILogger logger) +{ + public async Task 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 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 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(); + 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); + } +} diff --git a/server/QMax.Api/Services/MaxOutboxWorker.cs b/server/QMax.Api/Services/MaxOutboxWorker.cs new file mode 100644 index 0000000..cf965a3 --- /dev/null +++ b/server/QMax.Api/Services/MaxOutboxWorker.cs @@ -0,0 +1,41 @@ +using Microsoft.Extensions.Options; +using QMax.Api.Configuration; + +namespace QMax.Api.Services; + +public sealed class MaxOutboxWorker( + IServiceScopeFactory scopeFactory, + IOptions options, + ILogger 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(); + await outbox.ProcessOnceAsync(stoppingToken); + } + catch (Exception ex) + { + logger.LogWarning(ex, "QMAX outbox tick failed."); + } + } + } +} diff --git a/tests/QMax.Tests/ApiSmokeTests.cs b/tests/QMax.Tests/ApiSmokeTests.cs index 6519167..879023b 100644 --- a/tests/QMax.Tests/ApiSmokeTests.cs +++ b/tests/QMax.Tests/ApiSmokeTests.cs @@ -46,7 +46,8 @@ public sealed class ApiSmokeTests : IDisposable ["QMax:PairingCode"] = "test-pairing", ["QMax:JwtSecret"] = "development-only-qmax-secret-change-before-production", ["QMax:MaxMode"] = "Mock", - ["QMax:MaxPollIntervalSeconds"] = "600" + ["QMax:MaxPollIntervalSeconds"] = "600", + ["QMax:MaxOutboxPollIntervalSeconds"] = "600" }); }); }); @@ -89,7 +90,8 @@ public sealed class ApiSmokeTests : IDisposable var message = await messageResponse.Content.ReadFromJsonAsync(JsonOptions); Assert.NotNull(message); Assert.Equal("hello from test", message!.Text); - Assert.Equal(MessageDeliveryState.Sent, message.DeliveryState); + Assert.Null(message.ExternalId); + Assert.Equal(MessageDeliveryState.Sending, message.DeliveryState); var replyResponse = await client.PostAsJsonAsync( $"/api/chats/{chat.Id}/messages", @@ -98,6 +100,14 @@ public sealed class ApiSmokeTests : IDisposable var reply = await replyResponse.Content.ReadFromJsonAsync(JsonOptions); Assert.NotNull(reply); Assert.Equal(message.Id, reply!.ReplyToMessageId); + + Assert.True(await ProcessOutboxAsync(_factory) >= 2); + var sentMessages = await client.GetFromJsonAsync($"/api/chats/{chat.Id}/messages", JsonOptions); + Assert.All(sentMessages!, sent => + { + Assert.False(string.IsNullOrWhiteSpace(sent.ExternalId)); + Assert.Equal(MessageDeliveryState.Sent, sent.DeliveryState); + }); } [Fact] @@ -248,7 +258,7 @@ public sealed class ApiSmokeTests : IDisposable } [Fact] - public async Task UploadAttachmentFlowStoresDownloadsAndMarksSent() + public async Task UploadAttachmentFlowStoresDownloadsAndQueuesSend() { using var client = _factory.CreateClient(); await LoginAsync(client); @@ -278,7 +288,8 @@ public sealed class ApiSmokeTests : IDisposable await AssertStatusAsync(HttpStatusCode.OK, upload); var message = await upload.Content.ReadFromJsonAsync(JsonOptions); Assert.NotNull(message); - Assert.Equal(MessageDeliveryState.Sent, message!.DeliveryState); + Assert.Equal(MessageDeliveryState.Sending, message!.DeliveryState); + Assert.Null(message.ExternalId); Assert.Equal(baseMessage.Id, message.ReplyToMessageId); var attachment = Assert.Single(message.Attachments); Assert.Equal(AttachmentKind.Image, attachment.Kind); @@ -320,7 +331,8 @@ public sealed class ApiSmokeTests : IDisposable await AssertStatusAsync(HttpStatusCode.OK, upload); var sent = await upload.Content.ReadFromJsonAsync(JsonOptions); Assert.NotNull(sent); - Assert.Equal(bridge.MessageExternalId, sent!.ExternalId); + Assert.Null(sent!.ExternalId); + Assert.Equal(MessageDeliveryState.Sending, sent.DeliveryState); Assert.Single(sent.Attachments); var sync = await client.PostAsync("/api/max/sync", null); @@ -328,6 +340,7 @@ public sealed class ApiSmokeTests : IDisposable var messages = await client.GetFromJsonAsync($"/api/chats/{chat.Id}/messages", JsonOptions); var echoed = Assert.Single(messages!, message => message.ExternalId == bridge.MessageExternalId); + Assert.Equal(sent.Id, echoed.Id); var attachment = Assert.Single(echoed.Attachments); Assert.Equal("local-photo.png", attachment.FileName); @@ -367,7 +380,7 @@ public sealed class ApiSmokeTests : IDisposable Assert.Equal(target.Id, forwardedText!.ChatId); Assert.Equal("forward text marker", forwardedText.Text); Assert.Equal("You", forwardedText.ForwardedFrom); - Assert.Equal(MessageDeliveryState.Sent, forwardedText.DeliveryState); + Assert.Equal(MessageDeliveryState.Sending, forwardedText.DeliveryState); var bytes = new byte[] { 0x89, 0x50, 0x4E, 0x47, 0x44, 0x41, 0x54, 0x41 }; using var form = new MultipartFormDataContent(); @@ -389,7 +402,7 @@ public sealed class ApiSmokeTests : IDisposable Assert.Equal(target.Id, forwardedAttachmentMessage!.ChatId); Assert.Equal("forward caption", forwardedAttachmentMessage.Text); Assert.Equal("You", forwardedAttachmentMessage.ForwardedFrom); - Assert.Equal(MessageDeliveryState.Sent, forwardedAttachmentMessage.DeliveryState); + Assert.Equal(MessageDeliveryState.Sending, forwardedAttachmentMessage.DeliveryState); var forwardedAttachment = Assert.Single(forwardedAttachmentMessage.Attachments); Assert.Equal("forward.png", forwardedAttachment.FileName); Assert.Equal(AttachmentKind.Image, forwardedAttachment.Kind); @@ -518,7 +531,12 @@ public sealed class ApiSmokeTests : IDisposable await AssertStatusAsync(HttpStatusCode.OK, messageResponse); var message = await messageResponse.Content.ReadFromJsonAsync(JsonOptions); Assert.NotNull(message); + + Assert.True(await ProcessOutboxAsync(factory) >= 1); + var afterOutbox = await client.GetFromJsonAsync($"/api/chats/{chat.Id}/messages", JsonOptions); + message = Assert.Single(afterOutbox!, x => x.Id == message!.Id); Assert.False(string.IsNullOrWhiteSpace(message!.ExternalId)); + Assert.Equal(MessageDeliveryState.Sent, message.DeliveryState); using var editRequest = new HttpRequestMessage(HttpMethod.Patch, $"/api/chats/{chat.Id}/messages/{message.Id}") { @@ -549,7 +567,7 @@ public sealed class ApiSmokeTests : IDisposable } [Fact] - public async Task FailedMaxTextSendDoesNotCreateLocalMessage() + public async Task FailedMaxTextSendMarksLocalMessageFailed() { using var factory = _factory.WithWebHostBuilder(builder => { @@ -566,16 +584,23 @@ public sealed class ApiSmokeTests : IDisposable var response = await client.PostAsJsonAsync( $"/api/chats/{chat.Id}/messages", new SendMessageRequest("this must not be stored")); - await AssertStatusAsync(HttpStatusCode.BadGateway, response); + await AssertStatusAsync(HttpStatusCode.OK, response); + var queued = await response.Content.ReadFromJsonAsync(JsonOptions); + Assert.NotNull(queued); + Assert.Equal(MessageDeliveryState.Sending, queued!.DeliveryState); + Assert.True(await ProcessOutboxAsync(factory) >= 1); var messages = await client.GetFromJsonAsync($"/api/chats/{chat.Id}/messages", JsonOptions); Assert.NotNull(messages); - Assert.Empty(messages!); + var failed = Assert.Single(messages!, x => x.Id == queued.Id); + Assert.Equal(queued.Id, failed.Id); + Assert.Equal(MessageDeliveryState.Failed, failed.DeliveryState); + Assert.Contains("Simulated MAX send failure.", failed.Error); var chats = await client.GetFromJsonAsync("/api/chats", JsonOptions); var updated = Assert.Single(chats!, x => x.Id == chat.Id); - Assert.Null(updated.LastMessagePreview); - Assert.Null(updated.LastMessageAt); + Assert.Equal("this must not be stored", updated.LastMessagePreview); + Assert.NotNull(updated.LastMessageAt); } [Fact] @@ -1077,6 +1102,13 @@ public sealed class ApiSmokeTests : IDisposable return chat!; } + private static async Task ProcessOutboxAsync(WebApplicationFactory factory) + { + await using var scope = factory.Services.CreateAsyncScope(); + var outbox = scope.ServiceProvider.GetRequiredService(); + return await outbox.ProcessOnceAsync(CancellationToken.None); + } + private static async Task AssertStatusAsync(HttpStatusCode expected, HttpResponseMessage response) { if (response.StatusCode != expected) diff --git a/worker/src/server.js b/worker/src/server.js index d777e2e..155767f 100644 --- a/worker/src/server.js +++ b/worker/src/server.js @@ -14,9 +14,11 @@ const domStickerPrefix = "qmax-dom-sticker:"; const domEmojiPrefix = "qmax-dom-emoji:"; let context; +let contextInitialization; let page; let lastError = null; -let pageOperation = Promise.resolve(); +const pagesByRole = new Map(); +const pageOperations = new Map(); const networkLog = []; const app = express(); @@ -227,9 +229,9 @@ app.get("/media/fetch", async (req, res) => { } const media = await withPageLock(async () => { - const p = await ensurePage(); + const p = await ensurePage("incoming"); return await fetchMediaThroughMaxSession(p, mediaUrl); - }); + }, "incoming"); res.setHeader("Content-Type", media.contentType || "application/octet-stream"); res.setHeader("Content-Length", String(media.buffer.length)); @@ -242,8 +244,8 @@ app.get("/media/fetch", async (req, res) => { app.get("/updates", async (_req, res) => { try { const result = await withPageLock(async () => { - const p = await ensurePage(); - const currentStatus = await status(); + const p = await ensurePage("incoming"); + const currentStatus = await status(null, p); if (!currentStatus.isAuthorized) { return []; } @@ -251,7 +253,7 @@ app.get("/updates", async (_req, res) => { await clearSidebarSearch(p); await resetSidebarListToTop(p); return await extractUpdates(p); - }); + }, "incoming"); res.json(result); } catch (error) { res.status(500).json(errorStatus(error)); @@ -262,8 +264,8 @@ app.post("/chat/history", async (req, res) => { try { const externalChatId = String(req.body?.externalChatId || "").trim(); const result = await withPageLock(async () => { - const p = await ensurePage(); - const currentStatus = await status(); + const p = await ensurePage("incoming"); + const currentStatus = await status(null, p); if (!currentStatus.isAuthorized) { return null; } @@ -277,7 +279,7 @@ app.post("/chat/history", async (req, res) => { await scrollOpenChatToLatest(p); return await extractOpenChatHistory(p, externalChatId); - }); + }, "incoming"); if (!result) { return res.status(409).json({ error: "MAX is not authorized." }); @@ -293,8 +295,8 @@ app.post("/chat/clear-composer", async (req, res) => { try { const externalChatId = String(req.body?.externalChatId || "").trim(); const result = await withPageLock(async () => { - const p = await ensurePage(); - const currentStatus = await status(); + const p = await ensurePage("outgoing"); + const currentStatus = await status(null, p); if (!currentStatus.isAuthorized) { return { success: false, error: "MAX is not authorized.", composerText: null }; } @@ -308,7 +310,7 @@ app.post("/chat/clear-composer", async (req, res) => { const success = await clearComposerInput(p); return { success, error: success ? null : "MAX composer was not cleared.", composerText: await readComposerText(p) }; - }); + }, "outgoing"); res.json(result); } catch (error) { @@ -325,8 +327,8 @@ app.post("/chat/draft-text", async (req, res) => { } const result = await withPageLock(async () => { - const p = await ensurePage(); - const currentStatus = await status(); + const p = await ensurePage("outgoing"); + const currentStatus = await status(null, p); if (!currentStatus.isAuthorized) { return { success: false, error: "MAX is not authorized.", composerText: null }; } @@ -340,7 +342,7 @@ app.post("/chat/draft-text", async (req, res) => { const success = await fillComposerText(p, text, true); return { success, error: success ? null : "MAX composer was not filled.", composerText: await readComposerText(p) }; - }); + }, "outgoing"); res.json(result); } catch (error) { @@ -352,8 +354,8 @@ app.post("/chat/presence", async (req, res) => { try { const externalChatId = String(req.body?.externalChatId || "").trim(); const result = await withPageLock(async () => { - const p = await ensurePage(); - const currentStatus = await status(); + const p = await ensurePage("incoming"); + const currentStatus = await status(null, p); if (!currentStatus.isAuthorized) { return { isTyping: false, statusText: null, updatedAt: new Date().toISOString() }; } @@ -363,7 +365,7 @@ app.post("/chat/presence", async (req, res) => { } return await extractOpenChatPresence(p); - }); + }, "incoming"); res.json(result); } catch (error) { @@ -380,7 +382,7 @@ app.post("/send/text", async (req, res) => { } const result = await withPageLock(async () => { - const p = await ensurePage(); + const p = await ensurePage("outgoing"); if (externalChatId) { const opened = await ensureDomChatOpen(p, externalChatId, 8000, true); @@ -394,7 +396,7 @@ app.post("/send/text", async (req, res) => { return { success: false, externalMessageId: null, error: "MAX did not confirm the outgoing text message." }; } return { success: true, externalMessageId, error: null }; - }); + }, "outgoing"); res.json(result); } catch (error) { res.json({ success: false, externalMessageId: null, error: String(error?.message || error) }); @@ -411,7 +413,7 @@ app.post("/send/attachment", async (req, res) => { } const result = await withPageLock(async () => { - const p = await ensurePage(); + const p = await ensurePage("outgoing"); if (externalChatId) { const opened = await ensureDomChatOpen(p, externalChatId, 8000, true); if (!opened) { @@ -425,7 +427,7 @@ app.post("/send/attachment", async (req, res) => { } return { success: true, externalMessageId, error: null }; - }); + }, "outgoing"); res.json(result); } catch (error) { @@ -444,9 +446,9 @@ app.post("/message/edit", async (req, res) => { } const result = await withPageLock(async () => { - const p = await ensurePage(); + const p = await ensurePage("outgoing"); return await editMessageInMax(p, externalChatId, externalMessageId, currentText, text); - }); + }, "outgoing"); res.json(result); } catch (error) { res.json({ success: false, error: String(error?.message || error) }); @@ -463,9 +465,9 @@ app.post("/message/delete", async (req, res) => { } const result = await withPageLock(async () => { - const p = await ensurePage(); + const p = await ensurePage("outgoing"); return await deleteMessageInMax(p, externalChatId, externalMessageId, currentText); - }); + }, "outgoing"); res.json(result); } catch (error) { res.json({ success: false, error: String(error?.message || error) }); @@ -483,9 +485,9 @@ app.post("/message/reaction", async (req, res) => { } const result = await withPageLock(async () => { - const p = await ensurePage(); + const p = await ensurePage("outgoing"); return await setMessageReactionInMax(p, externalChatId, externalMessageId, currentText, emoji); - }); + }, "outgoing"); res.json(result); } catch (error) { res.json({ success: false, error: String(error?.message || error) }); @@ -503,9 +505,9 @@ app.delete("/message/reaction", async (req, res) => { } const result = await withPageLock(async () => { - const p = await ensurePage(); + const p = await ensurePage("outgoing"); return await clearMessageReactionInMax(p, externalChatId, externalMessageId, currentText, emoji); - }); + }, "outgoing"); res.json(result); } catch (error) { res.json({ success: false, error: String(error?.message || error) }); @@ -516,32 +518,70 @@ app.listen(port, () => { console.log(`QMAX MAX worker listening on ${port}, headless=${headless}`); }); -async function ensurePage() { - if (page) return page; - await cleanupChromiumProfileLocks(userDataDir); - context = await chromium.launchPersistentContext(userDataDir, { - headless, - acceptDownloads: true, - locale: "ru-RU", - timezoneId: process.env.MAX_TIMEZONE_ID || "Europe/Moscow", - viewport: { width: 1280, height: 900 }, - args: ["--disable-dev-shm-usage", "--no-sandbox"] - }); - await context.grantPermissions(["clipboard-read", "clipboard-write"], { origin: maxBaseUrl }).catch(() => {}); - page = context.pages()[0] || await context.newPage(); - attachNetworkCapture(page); - if (!page.url() || page.url() === "about:blank") { - await page.goto(maxBaseUrl, { waitUntil: "domcontentloaded", timeout: 60000 }); +async function ensurePage(role = "default") { + const normalizedRole = String(role || "default"); + const existing = pagesByRole.get(normalizedRole); + if (existing && !existing.isClosed()) return existing; + + await ensureContext(); + + if (normalizedRole === "default") { + if (!page || page.isClosed()) { + page = context.pages().find((candidate) => !candidate.isClosed()) || await context.newPage(); + pagesByRole.set("default", page); + attachNetworkCapture(page); + if (!page.url() || page.url() === "about:blank") { + await page.goto(maxBaseUrl, { waitUntil: "domcontentloaded", timeout: 60000 }); + } + } + + return page; } - return page; + + const rolePage = await context.newPage(); + pagesByRole.set(normalizedRole, rolePage); + attachNetworkCapture(rolePage); + await rolePage.goto(maxBaseUrl, { waitUntil: "domcontentloaded", timeout: 60000 }); + return rolePage; } -async function withPageLock(operation) { - const previousOperation = pageOperation; +async function ensureContext() { + if (context) return; + if (!contextInitialization) { + contextInitialization = (async () => { + await cleanupChromiumProfileLocks(userDataDir); + context = await chromium.launchPersistentContext(userDataDir, { + headless, + acceptDownloads: true, + locale: "ru-RU", + timezoneId: process.env.MAX_TIMEZONE_ID || "Europe/Moscow", + viewport: { width: 1280, height: 900 }, + args: ["--disable-dev-shm-usage", "--no-sandbox"] + }); + await context.grantPermissions(["clipboard-read", "clipboard-write"], { origin: maxBaseUrl }).catch(() => {}); + page = context.pages()[0] || await context.newPage(); + pagesByRole.set("default", page); + attachNetworkCapture(page); + if (!page.url() || page.url() === "about:blank") { + await page.goto(maxBaseUrl, { waitUntil: "domcontentloaded", timeout: 60000 }); + } + })().catch((error) => { + contextInitialization = null; + throw error; + }); + } + + await contextInitialization; +} + +async function withPageLock(operation, role = "default") { + const normalizedRole = String(role || "default"); + const previousOperation = pageOperations.get(normalizedRole) || Promise.resolve(); let release; - pageOperation = new Promise((resolve) => { + const nextOperation = new Promise((resolve) => { release = resolve; }); + pageOperations.set(normalizedRole, nextOperation); await previousOperation.catch(() => {}); try { @@ -551,9 +591,9 @@ async function withPageLock(operation) { } } -async function status(overrideStatus = null) { +async function status(overrideStatus = null, statusPage = null) { try { - const p = await ensurePage(); + const p = statusPage || await ensurePage(); const title = await safeTitle(p); const url = p.url(); const bodyText = await safeBodyText(p); @@ -575,17 +615,26 @@ async function status(overrideStatus = null) { function errorStatus(error) { lastError = String(error?.message || error); + const fallbackPage = firstKnownPage(); return { mode: "Worker", isAuthorized: false, status: "WorkerError", - url: page?.url?.() || null, + url: fallbackPage?.url?.() || null, title: null, lastError, updatedAt: new Date().toISOString() }; } +function firstKnownPage() { + return page || + pagesByRole.get("default") || + pagesByRole.get("incoming") || + pagesByRole.get("outgoing") || + null; +} + function isAllowedMediaUrl(value) { const raw = String(value || "").trim(); if (!raw) return false;