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

Queue chat bulk actions through SQL outbox

Browse Source
This commit is contained in:
sevenhill
2026-07-05 22:16:30 +03:00
parent f57faead53
commit dcc74252c6
10 changed files with 443 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/QMax.Api/Controllers/ChatsController.cs
+38 -35
View File
@@ -26,7 +26,9 @@ public sealed class ChatsController(
[HttpGet]
public async Task<ActionResult<IReadOnlyList<ChatDto>>> GetChats(CancellationToken cancellationToken)
{
var chats = (await db.Chats.ToArrayAsync(cancellationToken))
var chats = (await db.Chats
.Where(x => x.DeletedAt == null)
.ToArrayAsync(cancellationToken))
.OrderByDescending(x => x.IsPinned)
.ThenByDescending(x => x.UnreadCount > 0)
.ThenByDescending(x => x.LastMessageAt ?? x.UpdatedAt)
@@ -59,7 +61,7 @@ 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)
{
var chat = await db.Chats.FirstOrDefaultAsync(x => x.Id == chatId, cancellationToken);
var chat = await db.Chats.FirstOrDefaultAsync(x => x.Id == chatId && x.DeletedAt == null, cancellationToken);
if (chat is null)
{
return NotFound();
@@ -69,7 +71,7 @@ public sealed class ChatsController(
{
await syncService.SyncChatHistoryAsync(chat.ExternalId, chat.WebUrl, cancellationToken);
db.ChangeTracker.Clear();
chat = await db.Chats.FirstOrDefaultAsync(x => x.Id == chatId, cancellationToken);
chat = await db.Chats.FirstOrDefaultAsync(x => x.Id == chatId && x.DeletedAt == null, cancellationToken);
if (chat is null)
{
return NotFound();
@@ -104,7 +106,7 @@ public sealed class ChatsController(
[HttpPost("{chatId:guid}/read")]
public async Task<IActionResult> MarkRead(Guid chatId, CancellationToken cancellationToken)
{
var chat = await db.Chats.FirstOrDefaultAsync(x => x.Id == chatId, cancellationToken);
var chat = await db.Chats.FirstOrDefaultAsync(x => x.Id == chatId && x.DeletedAt == null, cancellationToken);
if (chat is null)
{
return NotFound();
@@ -136,19 +138,7 @@ public sealed class ChatsController(
return NotFound();
}
foreach (var chat in chats)
{
var maxFailure = await TryApplyMaxChatActionAsync(
chat,
(externalChatId, chatUrl) => maxBridge.ClearChatHistoryAsync(externalChatId, chatUrl, cancellationToken),
"clear history",
cancellationToken);
if (maxFailure is not null)
{
return maxFailure;
}
}
var now = DateTimeOffset.UtcNow;
var files = AttachmentFiles(chats).ToArray();
foreach (var chat in chats)
{
@@ -156,7 +146,12 @@ public sealed class ChatsController(
chat.LastMessageAt = null;
chat.LastMessagePreview = null;
chat.UnreadCount = 0;
chat.UpdatedAt = DateTimeOffset.UtcNow;
chat.HistoryClearedAt = now;
chat.UpdatedAt = now;
if (chat.DeletedAt is null)
{
QueuePendingMaxAction(chat, ChatPendingMaxAction.ClearHistory, now);
}
}
await db.SaveChangesAsync(cancellationToken);
@@ -180,21 +175,20 @@ public sealed class ChatsController(
return NotFound();
}
var now = DateTimeOffset.UtcNow;
var files = AttachmentFiles(chats).ToArray();
foreach (var chat in chats)
{
var maxFailure = await TryApplyMaxChatActionAsync(
chat,
(externalChatId, chatUrl) => maxBridge.DeleteChatAsync(externalChatId, chatUrl, cancellationToken),
"delete chat",
cancellationToken);
if (maxFailure is not null)
{
return maxFailure;
}
db.Messages.RemoveRange(chat.Messages);
chat.DeletedAt ??= now;
chat.HistoryClearedAt = now;
chat.LastMessageAt = null;
chat.LastMessagePreview = null;
chat.UnreadCount = 0;
chat.UpdatedAt = now;
QueuePendingMaxAction(chat, ChatPendingMaxAction.DeleteChat, now);
}
var files = AttachmentFiles(chats).ToArray();
db.Chats.RemoveRange(chats);
await db.SaveChangesAsync(cancellationToken);
DeleteFiles(files);
await hubContext.Clients.All.SendAsync("ChatListInvalidated", cancellationToken);
@@ -214,7 +208,7 @@ public sealed class ChatsController(
return BadRequest("Search query is required.");
}
var chatExists = await db.Chats.AnyAsync(x => x.Id == chatId, cancellationToken);
var chatExists = await db.Chats.AnyAsync(x => x.Id == chatId && x.DeletedAt == null, cancellationToken);
if (!chatExists)
{
return NotFound();
@@ -242,7 +236,7 @@ public sealed class ChatsController(
{
var chat = await db.Chats
.AsNoTracking()
.FirstOrDefaultAsync(x => x.Id == chatId, cancellationToken);
.FirstOrDefaultAsync(x => x.Id == chatId && x.DeletedAt == null, cancellationToken);
if (chat is null || string.IsNullOrWhiteSpace(chat.AvatarUrl))
{
@@ -263,7 +257,7 @@ public sealed class ChatsController(
{
var chat = await db.Chats
.AsNoTracking()
.FirstOrDefaultAsync(x => x.Id == chatId, cancellationToken);
.FirstOrDefaultAsync(x => x.Id == chatId && x.DeletedAt == null, cancellationToken);
if (chat is null)
{
@@ -285,7 +279,7 @@ public sealed class ChatsController(
[HttpPost("{chatId:guid}/messages")]
public async Task<ActionResult<MessageDto>> SendMessage(Guid chatId, SendMessageRequest request, CancellationToken cancellationToken)
{
var chat = await db.Chats.FindAsync([chatId], cancellationToken);
var chat = await db.Chats.FirstOrDefaultAsync(x => x.Id == chatId && x.DeletedAt == null, cancellationToken);
if (chat is null)
{
return NotFound();
@@ -527,7 +521,7 @@ public sealed class ChatsController(
return NotFound();
}
var target = await db.Chats.FindAsync([request.TargetChatId], cancellationToken);
var target = await db.Chats.FirstOrDefaultAsync(x => x.Id == request.TargetChatId && x.DeletedAt == null, cancellationToken);
if (target is null)
{
return NotFound();
@@ -600,7 +594,7 @@ public sealed class ChatsController(
[FromForm] Guid? replyToMessageId,
CancellationToken cancellationToken)
{
var chat = await db.Chats.FindAsync([chatId], cancellationToken);
var chat = await db.Chats.FirstOrDefaultAsync(x => x.Id == chatId && x.DeletedAt == null, cancellationToken);
if (chat is null)
{
return NotFound();
@@ -809,6 +803,15 @@ public sealed class ChatsController(
error?.Contains("menu was not opened", StringComparison.OrdinalIgnoreCase) == true;
}
private static void QueuePendingMaxAction(Chat chat, ChatPendingMaxAction action, DateTimeOffset requestedAt)
{
chat.PendingMaxAction = action;
chat.PendingMaxActionRequestedAt = requestedAt;
chat.PendingMaxActionLastAttemptAt = null;
chat.PendingMaxActionAttempts = 0;
chat.PendingMaxActionError = null;
}
private async Task<List<Chat>> LoadChatsWithAttachmentsAsync(IReadOnlyCollection<Guid> chatIds, CancellationToken cancellationToken)
{
return await db.Chats
server/QMax.Api/Data/Entities/Chat.cs
+7
View File
@@ -14,6 +14,13 @@ public sealed class Chat
public bool IsPinned { get; set; }
public bool IsMuted { get; set; }
public bool IsArchived { get; set; }
public DateTimeOffset? DeletedAt { get; set; }
public DateTimeOffset? HistoryClearedAt { get; set; }
public ChatPendingMaxAction? PendingMaxAction { get; set; }
public DateTimeOffset? PendingMaxActionRequestedAt { get; set; }
public DateTimeOffset? PendingMaxActionLastAttemptAt { get; set; }
public int PendingMaxActionAttempts { get; set; }
public string? PendingMaxActionError { get; set; }
public DateTimeOffset CreatedAt { get; set; } = DateTimeOffset.UtcNow;
public DateTimeOffset UpdatedAt { get; set; } = DateTimeOffset.UtcNow;
public ICollection<Message> Messages { get; set; } = new List<Message>();
server/QMax.Api/Data/Entities/Enums.cs
+6
View File
@@ -8,6 +8,12 @@ public enum ChatKind
MaxDialog = 3
}
public enum ChatPendingMaxAction
{
ClearHistory = 0,
DeleteChat = 1
}
public enum MessageDirection
{
Incoming = 0,
server/QMax.Api/Data/QMaxDbContext.cs
+6
View File
@@ -19,6 +19,8 @@ public sealed class QMaxDbContext(DbContextOptions<QMaxDbContext> options) : DbC
modelBuilder.Entity<User>().HasIndex(x => x.PhoneNumber);
modelBuilder.Entity<UserSession>().HasIndex(x => x.RefreshTokenHash).IsUnique();
modelBuilder.Entity<Chat>().HasIndex(x => x.ExternalId).IsUnique();
modelBuilder.Entity<Chat>().HasIndex(x => x.DeletedAt);
modelBuilder.Entity<Chat>().HasIndex(x => new { x.PendingMaxAction, x.PendingMaxActionRequestedAt });
modelBuilder.Entity<Message>().HasIndex(x => new { x.ChatId, x.SentAt });
modelBuilder.Entity<Message>().HasIndex(x => x.ExternalId);
modelBuilder.Entity<MessageAttachment>().HasIndex(x => x.StorageFileName).IsUnique();
@@ -30,6 +32,10 @@ public sealed class QMaxDbContext(DbContextOptions<QMaxDbContext> options) : DbC
.Property(x => x.Kind)
.HasConversion<string>();
modelBuilder.Entity<Chat>()
.Property(x => x.PendingMaxAction)
.HasConversion<string>();
modelBuilder.Entity<Message>()
.Property(x => x.Direction)
.HasConversion<string>();
server/QMax.Api/Program.cs
+45
View File
@@ -157,6 +157,41 @@ static async Task EnsureCompatibilitySchemaAsync(QMaxDbContext db)
await db.Database.ExecuteSqlRawAsync("ALTER TABLE Chats ADD COLUMN WebUrl TEXT;");
}
if (!chatColumns.Contains("DeletedAt"))
{
await db.Database.ExecuteSqlRawAsync("ALTER TABLE Chats ADD COLUMN DeletedAt TEXT;");
}
if (!chatColumns.Contains("HistoryClearedAt"))
{
await db.Database.ExecuteSqlRawAsync("ALTER TABLE Chats ADD COLUMN HistoryClearedAt TEXT;");
}
if (!chatColumns.Contains("PendingMaxAction"))
{
await db.Database.ExecuteSqlRawAsync("ALTER TABLE Chats ADD COLUMN PendingMaxAction TEXT;");
}
if (!chatColumns.Contains("PendingMaxActionRequestedAt"))
{
await db.Database.ExecuteSqlRawAsync("ALTER TABLE Chats ADD COLUMN PendingMaxActionRequestedAt TEXT;");
}
if (!chatColumns.Contains("PendingMaxActionLastAttemptAt"))
{
await db.Database.ExecuteSqlRawAsync("ALTER TABLE Chats ADD COLUMN PendingMaxActionLastAttemptAt TEXT;");
}
if (!chatColumns.Contains("PendingMaxActionAttempts"))
{
await db.Database.ExecuteSqlRawAsync("ALTER TABLE Chats ADD COLUMN PendingMaxActionAttempts INTEGER NOT NULL DEFAULT 0;");
}
if (!chatColumns.Contains("PendingMaxActionError"))
{
await db.Database.ExecuteSqlRawAsync("ALTER TABLE Chats ADD COLUMN PendingMaxActionError TEXT;");
}
var attachmentColumns = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
await using (var command = connection.CreateCommand())
{
@@ -189,6 +224,16 @@ static async Task EnsureCompatibilitySchemaAsync(QMaxDbContext db)
WHERE ExternalId IS NOT NULL AND ExternalId <> '';
""");
await db.Database.ExecuteSqlRawAsync("""
CREATE INDEX IF NOT EXISTS IX_Chats_DeletedAt
ON Chats (DeletedAt);
""");
await db.Database.ExecuteSqlRawAsync("""
CREATE INDEX IF NOT EXISTS IX_Chats_PendingMaxAction_PendingMaxActionRequestedAt
ON Chats (PendingMaxAction, PendingMaxActionRequestedAt);
""");
await db.Database.ExecuteSqlRawAsync("""
CREATE INDEX IF NOT EXISTS IX_Messages_ChatId_ExternalId
ON Messages (ChatId, ExternalId);
server/QMax.Api/Services/MaxBridgeSyncService.cs
+16
View File
@@ -89,6 +89,12 @@ public sealed class MaxBridgeSyncService(
}
}
if (chat?.DeletedAt is not null)
{
trackedChatsByExternalId[update.ExternalId] = chat;
continue;
}
var chatWasCreated = chat is null;
if (chat is null)
{
@@ -189,6 +195,11 @@ public sealed class MaxBridgeSyncService(
continue;
}
if (chat.HistoryClearedAt is { } historyClearedAt && incoming.SentAt <= historyClearedAt)
{
continue;
}
if (IsPresencePreview(incoming.Text) && (incoming.Attachments?.Count ?? 0) == 0)
{
continue;
@@ -347,6 +358,11 @@ public sealed class MaxBridgeSyncService(
var previousPreview = chat.LastMessagePreview;
var previousPreviewAt = chat.LastMessageAt;
var previewAt = ResolveListPreviewAt(update, previousPreviewAt ?? chat.UpdatedAt);
if (chat.HistoryClearedAt is { } historyClearedAt && previewAt <= historyClearedAt)
{
return false;
}
var samePreview = string.Equals(previousPreview, preview, StringComparison.Ordinal);
var samePreviewAt = previousPreviewAt is not null &&
Math.Abs((previousPreviewAt.Value - previewAt).TotalSeconds) < 1;
server/QMax.Api/Services/MaxOutboxService.cs
+145 -1
View File
@@ -18,6 +18,8 @@ public sealed class MaxOutboxService(
{
public async Task<int> ProcessOnceAsync(CancellationToken cancellationToken)
{
var processed = await ProcessPendingChatActionsAsync(cancellationToken);
var candidates = await db.Messages
.Include(message => message.Chat)
.Include(message => message.Attachments)
@@ -25,6 +27,8 @@ public sealed class MaxOutboxService(
.Where(message =>
message.Direction == MessageDirection.Outgoing &&
message.DeliveryState == MessageDeliveryState.Sending &&
message.Chat != null &&
message.Chat.DeletedAt == null &&
message.DeletedAt == null)
.ToListAsync(cancellationToken);
@@ -34,7 +38,6 @@ public sealed class MaxOutboxService(
.Take(10)
.ToArray();
var processed = 0;
foreach (var message in pending)
{
cancellationToken.ThrowIfCancellationRequested();
@@ -47,6 +50,147 @@ public sealed class MaxOutboxService(
return processed;
}
private async Task<int> ProcessPendingChatActionsAsync(CancellationToken cancellationToken)
{
var pending = await db.Chats
.Where(chat => chat.PendingMaxAction != null)
.ToListAsync(cancellationToken);
pending = pending
.OrderBy(chat => chat.PendingMaxActionRequestedAt ?? chat.UpdatedAt)
.ThenBy(chat => chat.Id)
.Take(10)
.ToList();
var processed = 0;
foreach (var chat in pending)
{
cancellationToken.ThrowIfCancellationRequested();
if (await ProcessChatActionAsync(chat, cancellationToken))
{
processed++;
}
}
return processed;
}
private async Task<bool> ProcessChatActionAsync(Chat chat, CancellationToken cancellationToken)
{
var pendingAction = chat.PendingMaxAction;
if (pendingAction is null)
{
return false;
}
MaxActionResult result;
try
{
result = pendingAction.Value switch
{
ChatPendingMaxAction.ClearHistory => await ApplyMaxChatActionAsync(
chat,
(externalChatId, chatUrl) => maxBridge.ClearChatHistoryAsync(externalChatId, chatUrl, cancellationToken),
cancellationToken),
ChatPendingMaxAction.DeleteChat => await ApplyMaxChatActionAsync(
chat,
(externalChatId, chatUrl) => maxBridge.DeleteChatAsync(externalChatId, chatUrl, cancellationToken),
cancellationToken),
_ => new MaxActionResult(false, $"Unsupported chat action: {pendingAction.Value}.")
};
}
catch (Exception ex)
{
logger.LogWarning(ex, "MAX chat action {Action} failed for chat {ChatId}.", pendingAction.Value, chat.Id);
result = new MaxActionResult(false, ex.Message);
}
var now = DateTimeOffset.UtcNow;
chat.PendingMaxActionLastAttemptAt = now;
if (result.Success)
{
chat.PendingMaxAction = null;
chat.PendingMaxActionRequestedAt = null;
chat.PendingMaxActionError = null;
}
else
{
chat.PendingMaxActionAttempts++;
chat.PendingMaxActionError = result.Error ?? $"MAX chat action {pendingAction.Value} failed.";
}
chat.UpdatedAt = now;
await db.SaveChangesAsync(cancellationToken);
if (!result.Success)
{
logger.LogWarning(
"MAX chat action {Action} failed for chat {ChatId}: {Error}",
pendingAction.Value,
chat.Id,
chat.PendingMaxActionError);
}
return true;
}
private async Task<MaxActionResult> ApplyMaxChatActionAsync(
Chat chat,
Func<string, string?, Task<MaxActionResult>> action,
CancellationToken cancellationToken)
{
if (string.IsNullOrWhiteSpace(chat.ExternalId))
{
return new MaxActionResult(true, null);
}
var chatUrl = chat.WebUrl;
if (string.IsNullOrWhiteSpace(chatUrl))
{
chatUrl = await ResolveAndPersistChatUrlAsync(chat, cancellationToken);
}
var result = await action(chat.ExternalId, chatUrl);
if (!result.Success && IsChatOpenFailure(result.Error))
{
var previousUrl = chatUrl;
chatUrl = await ResolveAndPersistChatUrlAsync(chat, cancellationToken);
if (!string.IsNullOrWhiteSpace(chatUrl) &&
!string.Equals(previousUrl, chatUrl, StringComparison.Ordinal))
{
result = await action(chat.ExternalId, chatUrl);
}
}
return result;
}
private async Task<string?> ResolveAndPersistChatUrlAsync(Chat chat, CancellationToken cancellationToken)
{
if (string.IsNullOrWhiteSpace(chat.ExternalId))
{
return chat.WebUrl;
}
var result = await maxBridge.ResolveChatUrlAsync(chat.ExternalId, cancellationToken);
if (!result.Success || string.IsNullOrWhiteSpace(result.ChatUrl))
{
return chat.WebUrl;
}
var nextWebUrl = result.ChatUrl.Trim();
if (!string.Equals(chat.WebUrl, nextWebUrl, StringComparison.Ordinal))
{
chat.WebUrl = nextWebUrl;
}
return chat.WebUrl;
}
private static bool IsChatOpenFailure(string? error)
{
return error?.Contains("not found or did not open", StringComparison.OrdinalIgnoreCase) == true ||
error?.Contains("menu was not opened", StringComparison.OrdinalIgnoreCase) == true;
}
private async Task<bool> ProcessMessageAsync(Message message, CancellationToken cancellationToken)
{
var chat = message.Chat;