sevenhill/QMAX
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Isolate MAX chat commands from outbox sends

Browse Source
This commit is contained in:
sevenhill
2026-07-05 22:48:05 +03:00
parent dcc74252c6
commit 0106130744
3 changed files with 143 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/QMax.Api/Services/MaxOutboxService.cs
+34 -3
View File
@@ -16,9 +16,12 @@ public sealed class MaxOutboxService(
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 ProcessPendingChatActionsAsync(cancellationToken);
var processed = 0;
var candidates = await db.Messages
.Include(message => message.Chat)
@@ -47,18 +50,20 @@ public sealed class MaxOutboxService(
}
}
return processed;
return processed + await ProcessPendingChatActionsAsync(cancellationToken);
}
private async Task<int> ProcessPendingChatActionsAsync(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)
.Take(1)
.ToList();
var processed = 0;
@@ -74,6 +79,30 @@ public sealed class MaxOutboxService(
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;
@@ -110,6 +139,8 @@ public sealed class MaxOutboxService(
{
chat.PendingMaxAction = null;
chat.PendingMaxActionRequestedAt = null;
chat.PendingMaxActionLastAttemptAt = null;
chat.PendingMaxActionAttempts = 0;
chat.PendingMaxActionError = null;
}
else
tests/QMax.Tests/ApiSmokeTests.cs
+91
View File
@@ -753,6 +753,78 @@ public sealed class ApiSmokeTests : IDisposable
}
}
[Fact]
public async Task OutboxSendsQueuedMessagesBeforePendingChatActions()
{
var bridge = new ResolvingDeleteMaxBridgeClient();
using var factory = _factory.WithWebHostBuilder(builder =>
{
builder.ConfigureTestServices(services =>
{
services.RemoveAll<IHostedService>();
services.RemoveAll<IMaxBridgeClient>();
services.AddSingleton<IMaxBridgeClient>(bridge);
});
});
using var client = factory.CreateClient();
await LoginAsync(client);
var deleteChat = await CreateDirectChatAsync(client, "delete-after-send-chat", "Delete After Send");
var deleteResponse = await client.PostAsJsonAsync(
"/api/chats/delete",
new ChatBulkActionRequest([deleteChat.Id]),
JsonOptions);
await AssertStatusAsync(HttpStatusCode.NoContent, deleteResponse);
var sendChat = await CreateDirectChatAsync(client, "send-before-delete-chat", "Send Before Delete");
var messageResponse = await client.PostAsJsonAsync(
$"/api/chats/{sendChat.Id}/messages",
new SendMessageRequest("send before chat action"),
JsonOptions);
await AssertStatusAsync(HttpStatusCode.OK, messageResponse);
Assert.Equal(2, await ProcessOutboxAsync(factory));
Assert.Equal(["send:text", "delete"], bridge.OperationOrder);
}
[Fact]
public async Task FailedPendingChatActionWaitsBeforeImmediateRetry()
{
var bridge = new ResolvingDeleteMaxBridgeClient { AlwaysFailDelete = true };
using var factory = _factory.WithWebHostBuilder(builder =>
{
builder.ConfigureTestServices(services =>
{
services.RemoveAll<IHostedService>();
services.RemoveAll<IMaxBridgeClient>();
services.AddSingleton<IMaxBridgeClient>(bridge);
});
});
using var client = factory.CreateClient();
await LoginAsync(client);
var chat = await CreateDirectChatAsync(client, "retry-delay-delete-chat", "Retry Delay Delete");
var deleteResponse = await client.PostAsJsonAsync(
"/api/chats/delete",
new ChatBulkActionRequest([chat.Id]),
JsonOptions);
await AssertStatusAsync(HttpStatusCode.NoContent, deleteResponse);
Assert.True(await ProcessOutboxAsync(factory) >= 1);
Assert.Single(bridge.DeleteChatUrls);
await ProcessOutboxAsync(factory);
Assert.Single(bridge.DeleteChatUrls);
await using var scope = factory.Services.CreateAsyncScope();
var db = scope.ServiceProvider.GetRequiredService<QMaxDbContext>();
var stored = await db.Chats.SingleAsync(x => x.Id == chat.Id);
Assert.Equal(ChatPendingMaxAction.DeleteChat, stored.PendingMaxAction);
Assert.Equal(1, stored.PendingMaxActionAttempts);
Assert.NotNull(stored.PendingMaxActionLastAttemptAt);
Assert.Equal("Simulated MAX delete failure.", stored.PendingMaxActionError);
}
[Fact]
public async Task BulkClearHistoryQueuesMaxActionAndClearsLocalMessages()
{
@@ -1662,6 +1734,10 @@ public sealed class ApiSmokeTests : IDisposable
public List<string?> DeleteChatUrls { get; } = [];
public string? ClearChatUrl { get; private set; }
public List<string?> ClearChatUrls { get; } = [];
public List<string> OperationOrder { get; } = [];
public List<string> SentTexts { get; } = [];
public bool AlwaysFailDelete { get; set; }
public bool AlwaysFailClear { get; set; }
public IReadOnlyList<MaxChatUpdate> Updates { get; set; } = [];
public Task<MaxBridgeStatus> GetStatusAsync(CancellationToken cancellationToken)
@@ -1707,8 +1783,14 @@ public sealed class ApiSmokeTests : IDisposable
public Task<MaxActionResult> ClearChatHistoryAsync(string externalChatId, string? chatUrl, CancellationToken cancellationToken)
{
OperationOrder.Add("clear");
ClearChatUrls.Add(chatUrl);
ClearChatUrl = chatUrl;
if (AlwaysFailClear)
{
return Task.FromResult(new MaxActionResult(false, "Simulated MAX clear failure."));
}
if (string.Equals(chatUrl, _staleUrlToReject, StringComparison.Ordinal))
{
return Task.FromResult(new MaxActionResult(false, "MAX chat menu was not opened."));
@@ -1721,8 +1803,14 @@ public sealed class ApiSmokeTests : IDisposable
public Task<MaxActionResult> DeleteChatAsync(string externalChatId, string? chatUrl, CancellationToken cancellationToken)
{
OperationOrder.Add("delete");
DeleteChatUrls.Add(chatUrl);
DeleteChatUrl = chatUrl;
if (AlwaysFailDelete)
{
return Task.FromResult(new MaxActionResult(false, "Simulated MAX delete failure."));
}
if (string.Equals(chatUrl, _staleUrlToReject, StringComparison.Ordinal))
{
return Task.FromResult(new MaxActionResult(false, "MAX chat menu was not opened."));
@@ -1735,11 +1823,14 @@ public sealed class ApiSmokeTests : IDisposable
public Task<MaxSendResult> SendTextAsync(string externalChatId, string? chatUrl, string text, CancellationToken cancellationToken)
{
OperationOrder.Add("send:text");
SentTexts.Add(text);
return Task.FromResult(new MaxSendResult(true, $"external-{Guid.NewGuid():N}", null));
}
public Task<MaxSendResult> SendAttachmentAsync(string externalChatId, string? chatUrl, string path, string caption, CancellationToken cancellationToken)
{
OperationOrder.Add("send:attachment");
return Task.FromResult(new MaxSendResult(true, $"external-file-{Guid.NewGuid():N}", null));
}
worker/src/server.js
+18 -8
View File
@@ -347,7 +347,7 @@ app.post("/chat/resolve-url", async (req, res) => {
}
const result = await withPageLock(async () => {
const p = await ensurePage("incoming");
const p = await ensurePage("commands");
const currentStatus = await status(null, p);
if (!currentStatus.isAuthorized) {
return { success: false, chatUrl: null, error: "MAX is not authorized." };
@@ -359,7 +359,7 @@ app.post("/chat/resolve-url", async (req, res) => {
chatUrl: chatUrl || null,
error: chatUrl ? null : "MAX chat URL was not found."
};
}, "incoming");
}, "commands");
res.json(result);
} catch (error) {
@@ -376,9 +376,9 @@ app.post("/chat/clear-history", async (req, res) => {
}
const result = await withPageLock(async () => {
const p = await ensurePage("incoming");
const p = await ensurePage("commands");
return await clearChatHistoryInMax(p, externalChatId, chatUrl);
}, "incoming");
}, "commands");
res.json(result);
} catch (error) {
res.json({ success: false, error: String(error?.message || error) });
@@ -394,9 +394,9 @@ app.post("/chat/delete", async (req, res) => {
}
const result = await withPageLock(async () => {
const p = await ensurePage("incoming");
const p = await ensurePage("commands");
return await deleteChatInMax(p, externalChatId, chatUrl);
}, "incoming");
}, "commands");
res.json(result);
} catch (error) {
res.json({ success: false, error: String(error?.message || error) });
@@ -416,7 +416,7 @@ app.post("/chat/menu-probe", async (req, res) => {
}
const result = await withPageLock(async () => {
const p = await ensurePage("incoming");
const p = await ensurePage("commands");
const menuOpened = await openSidebarChatMenu(p, externalChatId, chatUrl, labels);
const actionClicked = menuOpened && probeAction
? await clickActionByTextWithRetry(p, labels, 1500)
@@ -434,7 +434,7 @@ app.post("/chat/menu-probe", async (req, res) => {
actionClicked,
visibleActions
};
}, "incoming");
}, "commands");
res.json(result);
} catch (error) {
res.json({ success: false, error: String(error?.message || error), chatUrl: null });
@@ -759,6 +759,7 @@ async function status(overrideStatus = null, statusPage = null) {
url,
title,
lastError,
pageRoles: getOpenPageRoles(),
updatedAt: new Date().toISOString()
};
} catch (error) {
@@ -776,6 +777,7 @@ function errorStatus(error) {
url: fallbackPage?.url?.() || null,
title: null,
lastError,
pageRoles: getOpenPageRoles(),
updatedAt: new Date().toISOString()
};
}
@@ -785,9 +787,17 @@ function firstKnownPage() {
pagesByRole.get("default") ||
pagesByRole.get("incoming") ||
pagesByRole.get("outgoing") ||
pagesByRole.get("commands") ||
null;
}
function getOpenPageRoles() {
return Array.from(pagesByRole.entries())
.filter(([, rolePage]) => rolePage && !rolePage.isClosed())
.map(([roleName]) => roleName)
.sort();
}
function isAllowedMediaUrl(value) {
const raw = String(value || "").trim();
if (!raw) return false;