From 3bc6456ed742327a00299de3ee211ac55b046b6e Mon Sep 17 00:00:00 2001 From: sevenhill Date: Sun, 5 Jul 2026 23:20:20 +0300 Subject: [PATCH] Split MAX bot loops and make chats cache-first --- .../xyz/kusoft/qmax/core/network/QMaxApi.kt | 2 +- .../java/xyz/kusoft/qmax/ui/QMaxViewModel.kt | 61 ++++++++++++++++++- .../QMax.Api/Controllers/ChatsController.cs | 9 ++- server/QMax.Api/Services/MaxOutboxService.cs | 11 +++- server/QMax.Api/Services/MaxOutboxWorker.cs | 24 +++++++- tests/QMax.Tests/ApiSmokeTests.cs | 51 ++++++++++++++-- worker/src/server.js | 15 ++++- 7 files changed, 157 insertions(+), 16 deletions(-) diff --git a/android/app/src/main/java/xyz/kusoft/qmax/core/network/QMaxApi.kt b/android/app/src/main/java/xyz/kusoft/qmax/core/network/QMaxApi.kt index ca97758..085a611 100644 --- a/android/app/src/main/java/xyz/kusoft/qmax/core/network/QMaxApi.kt +++ b/android/app/src/main/java/xyz/kusoft/qmax/core/network/QMaxApi.kt @@ -80,7 +80,7 @@ class QMaxApi { } suspend fun messages(sessionServerUrl: String, token: String, chatId: String): List { - return get(sessionServerUrl, "/api/chats/$chatId/messages", token) + return get(sessionServerUrl, "/api/chats/$chatId/messages?sync=false", token) } suspend fun markChatRead(sessionServerUrl: String, token: String, chatId: String) { diff --git a/android/app/src/main/java/xyz/kusoft/qmax/ui/QMaxViewModel.kt b/android/app/src/main/java/xyz/kusoft/qmax/ui/QMaxViewModel.kt index 76dc453..31ff8cd 100644 --- a/android/app/src/main/java/xyz/kusoft/qmax/ui/QMaxViewModel.kt +++ b/android/app/src/main/java/xyz/kusoft/qmax/ui/QMaxViewModel.kt @@ -23,6 +23,8 @@ import xyz.kusoft.qmax.core.model.QMaxSession import xyz.kusoft.qmax.core.network.QMaxHttpException import xyz.kusoft.qmax.core.realtime.QMaxRealtimeClient import java.io.File +import java.time.OffsetDateTime +import java.util.UUID data class QMaxUiState( val serverUrl: String = BuildConfig.QMAX_DEFAULT_SERVER_URL, @@ -328,7 +330,7 @@ class QMaxViewModel(private val repository: QMaxRepository) : ViewModel() { } realtime.joinChat(chat.id) } - loadMessages(showLoading = true) + loadMessages(showLoading = false) presencePollJob?.cancel() presencePollJob = viewModelScope.launch { while (state.value.selectedChat?.id == chat.id) { @@ -574,7 +576,45 @@ class QMaxViewModel(private val repository: QMaxRepository) : ViewModel() { val sentMessages = mutableListOf() if (attachmentUris.isEmpty()) { - sentMessages += repository.sendMessage(session, chat.id, text, replyToMessageId) + val pending = pendingTextMessage(chat.id, text, replyToMessageId) + repository.cacheRealtimeMessage(session, pending) + repository.clearDraft(chat.id) + val current = state.value + if (current.selectedChat?.id == chat.id) { + state.value = current.copy( + composerText = "", + replyTarget = null, + messages = upsertMessage(current.messages, pending) + ) + } + onSuccess() + + try { + val sent = repository.sendMessage(session, chat.id, text, replyToMessageId) + repository.deleteCachedMessage(session, chat.id, pending.id) + repository.cacheRealtimeMessage(session, sent) + val afterSend = state.value + if (afterSend.selectedChat?.id == chat.id) { + state.value = afterSend.copy( + messages = replaceMessage(afterSend.messages, pending.id, sent) + ) + } + loadChats() + } catch (error: Throwable) { + val failed = pending.copy( + deliveryState = "Failed", + error = error.message ?: "\u041e\u0448\u0438\u0431\u043a\u0430 \u043e\u0442\u043f\u0440\u0430\u0432\u043a\u0438" + ) + repository.cacheRealtimeMessage(session, failed) + val afterFailure = state.value + if (afterFailure.selectedChat?.id == chat.id) { + state.value = afterFailure.copy( + messages = replaceMessage(afterFailure.messages, pending.id, failed) + ) + } + throw error + } + return@launchLoading } else { attachmentUris.forEachIndexed { index, uri -> val caption = text.takeIf { index == 0 && it.isNotBlank() } @@ -1101,6 +1141,23 @@ class QMaxViewModel(private val repository: QMaxRepository) : ViewModel() { return (messages.filterNot { it.id == message.id } + message).sortedBy { it.sentAt } } + private fun replaceMessage(messages: List, oldMessageId: String, message: MessageDto): List { + return (messages.filterNot { it.id == oldMessageId || it.id == message.id } + message).sortedBy { it.sentAt } + } + + private fun pendingTextMessage(chatId: String, text: String, replyToMessageId: String?): MessageDto { + return MessageDto( + id = "local-${UUID.randomUUID()}", + chatId = chatId, + senderName = "You", + direction = "Outgoing", + text = text, + sentAt = OffsetDateTime.now().toString(), + replyToMessageId = replyToMessageId, + deliveryState = "Sending" + ) + } + private fun normalizeChats( chats: List, selectedChatId: String? = state.value.selectedChat?.id diff --git a/server/QMax.Api/Controllers/ChatsController.cs b/server/QMax.Api/Controllers/ChatsController.cs index 7cf77f3..19032aa 100644 --- a/server/QMax.Api/Controllers/ChatsController.cs +++ b/server/QMax.Api/Controllers/ChatsController.cs @@ -59,7 +59,12 @@ public sealed class ChatsController( } [HttpGet("{chatId:guid}/messages")] - public async Task>> GetMessages(Guid chatId, int take = 80, DateTimeOffset? before = null, CancellationToken cancellationToken = default) + public async Task>> GetMessages( + Guid chatId, + int take = 80, + DateTimeOffset? before = null, + bool sync = false, + CancellationToken cancellationToken = default) { var chat = await db.Chats.FirstOrDefaultAsync(x => x.Id == chatId && x.DeletedAt == null, cancellationToken); if (chat is null) @@ -67,7 +72,7 @@ public sealed class ChatsController( return NotFound(); } - if (before is null && !string.IsNullOrWhiteSpace(chat.ExternalId)) + if (sync && before is null && !string.IsNullOrWhiteSpace(chat.ExternalId)) { await syncService.SyncChatHistoryAsync(chat.ExternalId, chat.WebUrl, cancellationToken); db.ChangeTracker.Clear(); diff --git a/server/QMax.Api/Services/MaxOutboxService.cs b/server/QMax.Api/Services/MaxOutboxService.cs index 5ba2b0b..46c638d 100644 --- a/server/QMax.Api/Services/MaxOutboxService.cs +++ b/server/QMax.Api/Services/MaxOutboxService.cs @@ -21,8 +21,12 @@ public sealed class MaxOutboxService( public async Task ProcessOnceAsync(CancellationToken cancellationToken) { - var processed = 0; + var processed = await ProcessPendingMessagesOnceAsync(cancellationToken); + return processed + await ProcessPendingChatActionsOnceAsync(cancellationToken); + } + public async Task ProcessPendingMessagesOnceAsync(CancellationToken cancellationToken) + { var candidates = await db.Messages .Include(message => message.Chat) .Include(message => message.Attachments) @@ -41,6 +45,7 @@ public sealed class MaxOutboxService( .Take(10) .ToArray(); + var processed = 0; foreach (var message in pending) { cancellationToken.ThrowIfCancellationRequested(); @@ -50,10 +55,10 @@ public sealed class MaxOutboxService( } } - return processed + await ProcessPendingChatActionsAsync(cancellationToken); + return processed; } - private async Task ProcessPendingChatActionsAsync(CancellationToken cancellationToken) + public async Task ProcessPendingChatActionsOnceAsync(CancellationToken cancellationToken) { var now = DateTimeOffset.UtcNow; var pending = await db.Chats diff --git a/server/QMax.Api/Services/MaxOutboxWorker.cs b/server/QMax.Api/Services/MaxOutboxWorker.cs index cf965a3..2f39efe 100644 --- a/server/QMax.Api/Services/MaxOutboxWorker.cs +++ b/server/QMax.Api/Services/MaxOutboxWorker.cs @@ -13,6 +13,26 @@ public sealed class MaxOutboxWorker( protected override async Task ExecuteAsync(CancellationToken stoppingToken) { var delay = TimeSpan.FromSeconds(Math.Max(1, _options.MaxOutboxPollIntervalSeconds)); + + await Task.WhenAll( + RunOutboxLoopAsync( + "message outbox", + delay, + (outbox, cancellationToken) => outbox.ProcessPendingMessagesOnceAsync(cancellationToken), + stoppingToken), + RunOutboxLoopAsync( + "chat command outbox", + delay, + (outbox, cancellationToken) => outbox.ProcessPendingChatActionsOnceAsync(cancellationToken), + stoppingToken)); + } + + private async Task RunOutboxLoopAsync( + string workerName, + TimeSpan delay, + Func> process, + CancellationToken stoppingToken) + { using var timer = new PeriodicTimer(delay); while (!stoppingToken.IsCancellationRequested) @@ -30,11 +50,11 @@ public sealed class MaxOutboxWorker( { await using var scope = scopeFactory.CreateAsyncScope(); var outbox = scope.ServiceProvider.GetRequiredService(); - await outbox.ProcessOnceAsync(stoppingToken); + await process(outbox, stoppingToken); } catch (Exception ex) { - logger.LogWarning(ex, "QMAX outbox tick failed."); + logger.LogWarning(ex, "QMAX {WorkerName} tick failed.", workerName); } } } diff --git a/tests/QMax.Tests/ApiSmokeTests.cs b/tests/QMax.Tests/ApiSmokeTests.cs index 24cc3e1..3b7eb42 100644 --- a/tests/QMax.Tests/ApiSmokeTests.cs +++ b/tests/QMax.Tests/ApiSmokeTests.cs @@ -602,8 +602,9 @@ public sealed class ApiSmokeTests : IDisposable { using var factory = _factory.WithWebHostBuilder(builder => { - builder.ConfigureServices(services => + builder.ConfigureTestServices(services => { + services.RemoveAll(); services.RemoveAll(); services.AddSingleton(); }); @@ -659,8 +660,9 @@ public sealed class ApiSmokeTests : IDisposable var bridge = new ResolvingDeleteMaxBridgeClient(); using var factory = _factory.WithWebHostBuilder(builder => { - builder.ConfigureServices(services => + builder.ConfigureTestServices(services => { + services.RemoveAll(); services.RemoveAll(); services.AddSingleton(bridge); }); @@ -703,6 +705,40 @@ public sealed class ApiSmokeTests : IDisposable } } + [Fact] + public async Task GetMessagesReturnsSqlCacheWithoutInlineMaxHistorySyncByDefault() + { + var bridge = new ResolvingDeleteMaxBridgeClient(); + using var factory = _factory.WithWebHostBuilder(builder => + { + builder.ConfigureTestServices(services => + { + services.RemoveAll(); + services.RemoveAll(); + services.AddSingleton(bridge); + }); + }); + using var client = factory.CreateClient(); + await LoginAsync(client); + + var chat = await CreateDirectChatAsync(client, "cache-first-chat", "Cache First"); + var messageResponse = await client.PostAsJsonAsync( + $"/api/chats/{chat.Id}/messages", + new SendMessageRequest("cached sql message"), + JsonOptions); + await AssertStatusAsync(HttpStatusCode.OK, messageResponse); + + var messages = await client.GetFromJsonAsync($"/api/chats/{chat.Id}/messages", JsonOptions); + + Assert.Contains(messages!, x => x.Text == "cached sql message"); + Assert.Equal(0, bridge.HistoryFetchCount); + + messages = await client.GetFromJsonAsync($"/api/chats/{chat.Id}/messages?sync=true", JsonOptions); + + Assert.Contains(messages!, x => x.Text == "cached sql message"); + Assert.Equal(1, bridge.HistoryFetchCount); + } + [Fact] public async Task BulkDeleteOutboxRefreshesStaleChatUrlAfterMenuOpenFailure() { @@ -711,8 +747,9 @@ public sealed class ApiSmokeTests : IDisposable var bridge = new ResolvingDeleteMaxBridgeClient(staleUrl); using var factory = _factory.WithWebHostBuilder(builder => { - builder.ConfigureServices(services => + builder.ConfigureTestServices(services => { + services.RemoveAll(); services.RemoveAll(); services.AddSingleton(bridge); }); @@ -832,8 +869,9 @@ public sealed class ApiSmokeTests : IDisposable var bridge = new ResolvingDeleteMaxBridgeClient(); using var factory = _factory.WithWebHostBuilder(builder => { - builder.ConfigureServices(services => + builder.ConfigureTestServices(services => { + services.RemoveAll(); services.RemoveAll(); services.AddSingleton(bridge); }); @@ -952,8 +990,9 @@ public sealed class ApiSmokeTests : IDisposable { using var factory = _factory.WithWebHostBuilder(builder => { - builder.ConfigureServices(services => + builder.ConfigureTestServices(services => { + services.RemoveAll(); services.RemoveAll(); services.AddSingleton(); }); @@ -1738,6 +1777,7 @@ public sealed class ApiSmokeTests : IDisposable public List SentTexts { get; } = []; public bool AlwaysFailDelete { get; set; } public bool AlwaysFailClear { get; set; } + public int HistoryFetchCount { get; private set; } public IReadOnlyList Updates { get; set; } = []; public Task GetStatusAsync(CancellationToken cancellationToken) @@ -1767,6 +1807,7 @@ public sealed class ApiSmokeTests : IDisposable public Task FetchChatHistoryAsync(string externalChatId, string? chatUrl, CancellationToken cancellationToken) { + HistoryFetchCount++; return Task.FromResult(null); } diff --git a/worker/src/server.js b/worker/src/server.js index 7df7c1e..2b03ba6 100644 --- a/worker/src/server.js +++ b/worker/src/server.js @@ -674,7 +674,10 @@ app.listen(port, () => { async function ensurePage(role = "default") { const normalizedRole = String(role || "default"); const existing = pagesByRole.get(normalizedRole); - if (existing && !existing.isClosed()) return existing; + if (existing && !existing.isClosed()) { + await waitForRolePageReady(existing, normalizedRole); + return existing; + } await ensureContext(); @@ -695,9 +698,19 @@ async function ensurePage(role = "default") { pagesByRole.set(normalizedRole, rolePage); attachNetworkCapture(rolePage); await rolePage.goto(maxBaseUrl, { waitUntil: "domcontentloaded", timeout: 60000 }); + await waitForRolePageReady(rolePage, normalizedRole); return rolePage; } +async function waitForRolePageReady(rolePage, role) { + if (role === "default") { + return; + } + + await rolePage.waitForLoadState("domcontentloaded", { timeout: 15000 }).catch(() => {}); + await waitForLoginStage(rolePage, "Authorized", 15000).catch(() => false); +} + async function ensureContext() { if (context) return; if (!contextInitialization) {