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

Split MAX bot loops and make chats cache-first

Browse Source
This commit is contained in:
sevenhill
2026-07-05 23:20:20 +03:00
parent 0106130744
commit 3bc6456ed7
7 changed files with 157 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
android/app/src/main/java/xyz/kusoft/qmax/core/network/QMaxApi.kt
+1 -1
View File
@@ -80,7 +80,7 @@ class QMaxApi {
}
suspend fun messages(sessionServerUrl: String, token: String, chatId: String): List<MessageDto> {
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) {
android/app/src/main/java/xyz/kusoft/qmax/ui/QMaxViewModel.kt
+59 -2
View File
@@ -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<MessageDto>()
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<MessageDto>, oldMessageId: String, message: MessageDto): List<MessageDto> {
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<ChatDto>,
selectedChatId: String? = state.value.selectedChat?.id
server/QMax.Api/Controllers/ChatsController.cs
+7 -2
View File
@@ -59,7 +59,12 @@ public sealed class ChatsController(
}
[HttpGet("{chatId:guid}/messages")]
public async Task<ActionResult<IReadOnlyList<MessageDto>>> GetMessages(Guid chatId, int take = 80, DateTimeOffset? before = null, CancellationToken cancellationToken = default)
public async Task<ActionResult<IReadOnlyList<MessageDto>>> 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();
server/QMax.Api/Services/MaxOutboxService.cs
+8 -3
View File
@@ -21,8 +21,12 @@ public sealed class MaxOutboxService(
public async Task<int> ProcessOnceAsync(CancellationToken cancellationToken)
{
var processed = 0;
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)
@@ -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<int> ProcessPendingChatActionsAsync(CancellationToken cancellationToken)
public async Task<int> ProcessPendingChatActionsOnceAsync(CancellationToken cancellationToken)
{
var now = DateTimeOffset.UtcNow;
var pending = await db.Chats
server/QMax.Api/Services/MaxOutboxWorker.cs
+22 -2
View File
@@ -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<MaxOutboxService, CancellationToken, Task<int>> 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<MaxOutboxService>();
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);
}
}
}
tests/QMax.Tests/ApiSmokeTests.cs
+46 -5
View File
@@ -602,8 +602,9 @@ public sealed class ApiSmokeTests : IDisposable
{
using var factory = _factory.WithWebHostBuilder(builder =>
{
builder.ConfigureServices(services =>
builder.ConfigureTestServices(services =>
{
services.RemoveAll<IHostedService>();
services.RemoveAll<IMaxBridgeClient>();
services.AddSingleton<IMaxBridgeClient, FailingActionMaxBridgeClient>();
});
@@ -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<IHostedService>();
services.RemoveAll<IMaxBridgeClient>();
services.AddSingleton<IMaxBridgeClient>(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<IHostedService>();
services.RemoveAll<IMaxBridgeClient>();
services.AddSingleton<IMaxBridgeClient>(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<MessageDto[]>($"/api/chats/{chat.Id}/messages", JsonOptions);
Assert.Contains(messages!, x => x.Text == "cached sql message");
Assert.Equal(0, bridge.HistoryFetchCount);
messages = await client.GetFromJsonAsync<MessageDto[]>($"/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<IHostedService>();
services.RemoveAll<IMaxBridgeClient>();
services.AddSingleton<IMaxBridgeClient>(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<IHostedService>();
services.RemoveAll<IMaxBridgeClient>();
services.AddSingleton<IMaxBridgeClient>(bridge);
});
@@ -952,8 +990,9 @@ public sealed class ApiSmokeTests : IDisposable
{
using var factory = _factory.WithWebHostBuilder(builder =>
{
builder.ConfigureServices(services =>
builder.ConfigureTestServices(services =>
{
services.RemoveAll<IHostedService>();
services.RemoveAll<IMaxBridgeClient>();
services.AddSingleton<IMaxBridgeClient, FailingSendMaxBridgeClient>();
});
@@ -1738,6 +1777,7 @@ public sealed class ApiSmokeTests : IDisposable
public List<string> SentTexts { get; } = [];
public bool AlwaysFailDelete { get; set; }
public bool AlwaysFailClear { get; set; }
public int HistoryFetchCount { get; private set; }
public IReadOnlyList<MaxChatUpdate> Updates { get; set; } = [];
public Task<MaxBridgeStatus> GetStatusAsync(CancellationToken cancellationToken)
@@ -1767,6 +1807,7 @@ public sealed class ApiSmokeTests : IDisposable
public Task<MaxChatUpdate?> FetchChatHistoryAsync(string externalChatId, string? chatUrl, CancellationToken cancellationToken)
{
HistoryFetchCount++;
return Task.FromResult<MaxChatUpdate?>(null);
}
worker/src/server.js
+14 -1
View File
@@ -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) {