From 9827b5df33b0c6f9200259ffc7e06afb555afc01 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=98=D0=BB=D1=8C=D1=8F=D1=81=20=D0=A1=D1=83=D0=BB=D1=82?= =?UTF-8?q?=D0=B0=D0=BD=D0=BE=D0=B2?= Date: Tue, 29 Sep 2026 20:31:59 +0300 Subject: [PATCH] =?UTF-8?q?fix(chat):=20cross-tab=20lock=20=D1=80=D0=B5?= =?UTF-8?q?=D0=BF=D0=BB=D0=B5=D1=8F=20=D0=BE=D1=84=D0=BB=D0=B0=D0=B9=D0=BD?= =?UTF-8?q?-=D0=BE=D1=87=D0=B5=D1=80=D0=B5=D0=B4=D0=B8=20+=20=D0=B8=D0=B4?= =?UTF-8?q?=D0=B5=D0=BC=D0=BF=D0=BE=D1=82=D0=B5=D0=BD=D1=82=D0=BD=D0=BE?= =?UTF-8?q?=D1=81=D1=82=D1=8C=20=D0=BC=D0=B5=D1=81=D1=81=D0=B5=D0=BD=D0=B4?= =?UTF-8?q?=D0=B6=D0=B5=D1=80=D0=B0=20(conv=5Fmessages)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Клиентская часть гонки (вариант B): Web Locks API (navigator.locks) — кросс-вкладочный мьютекс iistwin-offline-sync-replay на replay-проход в useOfflineSync: первая вкладка под локом реплеит и удаляет записи, остальные видят пустую очередь. Fallback для сред без Web Locks — старое поведение (дубли гасит серверная идемпотентность). Race-тест: без locks → 2 POST, с locks → 1 POST для одной записи очереди. Мессенджер (conv_messages) — та же защита, что у task_messages: - conversation_messages.client_message_id + partial unique (миграция 0087); - POST /api/messenger/conversations/:id/messages: повтор с тем же clientMessageId → существующее сообщение без insert и без SSE/уведомлений, гонка insert'ов ловится по 23505; - useChatController шлёт crypto.randomUUID(), attachment-реплей мессенджера тоже использует id записи очереди. --- .../src/components/chat/useChatController.ts | 5 +- client/src/hooks/useOfflineSync.ts | 21 ++++-- migrations/0087_conv_message_idempotency.sql | 10 +++ server/routes/messenger.messages.routes.ts | 66 +++++++++++++++---- shared/schema.ts | 3 + tests/offline-queue-multitab-race.test.tsx | 58 ++++++++++++++-- 6 files changed, 141 insertions(+), 22 deletions(-) create mode 100644 migrations/0087_conv_message_idempotency.sql diff --git a/client/src/components/chat/useChatController.ts b/client/src/components/chat/useChatController.ts index 2cc0b1b..53591a9 100644 --- a/client/src/components/chat/useChatController.ts +++ b/client/src/components/chat/useChatController.ts @@ -964,7 +964,7 @@ export function useChatController() { }; const sendMutation = useMutation({ - mutationFn: async (payload: { message: string; replyToId?: number; mentionedUserIds?: number[]; attachments?: ChatAttachment[] }) => { + mutationFn: async (payload: { message: string; replyToId?: number; mentionedUserIds?: number[]; attachments?: ChatAttachment[]; clientMessageId?: string }) => { const res = await apiRequest('POST', `/api/messenger/conversations/${activeConvId}/messages`, payload); const data = await res.json(); if (!data.success) throw new Error(data.error); @@ -1180,6 +1180,9 @@ export function useChatController() { replyToId: replyTo?.id, mentionedUserIds: mentionedUserIds.length > 0 ? mentionedUserIds : undefined, attachments: undefined, + // Ключ идемпотентности: реплеи офлайн-очереди (в т.ч. из разных вкладок) + // идут с одним ключом → серверный дедуп по client_message_id (миграция 0087) + clientMessageId: crypto.randomUUID(), }); setMentionedUserIds([]); setMentionSearch(null); diff --git a/client/src/hooks/useOfflineSync.ts b/client/src/hooks/useOfflineSync.ts index 5ea174d..c0d168a 100644 --- a/client/src/hooks/useOfflineSync.ts +++ b/client/src/hooks/useOfflineSync.ts @@ -86,6 +86,17 @@ function updateAppBadge(n: number) { } } +// Кросс-вкладочный мьютекс на реплей офлайн-очереди (Web Locks API). Без него N +// открытых вкладок реплеят одну и ту же IndexedDB-очередь НЕЗАВИСИМО (модульный +// _syncLock свой у каждой вкладки) → дубли сообщений/мутаций на сервере. +// Fallback для сред без Web Locks — старое поведение (дубли гасит серверная +// идемпотентность по clientMessageId, миграции 0086/0087). +async function withCrossTabSyncLock(fn: () => Promise): Promise { + const locks = (navigator as unknown as { locks?: { request: (name: string, cb: () => Promise) => Promise } }).locks; + if (!locks?.request) return fn(); + return locks.request('iistwin-offline-sync-replay', fn); +} + async function replayRequest(item: OfflineQueueItem): Promise { const headers: Record = { 'Content-Type': 'application/json', @@ -257,10 +268,8 @@ async function trySendAttachmentMessage( }; // Идемпотентность: id записи очереди стабилен между реплеями (в т.ч. из разных - // вкладок) — сервер дедупит повторы по client_message_id (миграция 0086). - if (item.type === 'task') { - payload.clientMessageId = (item as { clientMessageId?: string }).clientMessageId ?? item.id; - } + // вкладок) — сервер дедупит повторы по client_message_id (миграции 0086/0087). + payload.clientMessageId = (item as { clientMessageId?: string }).clientMessageId ?? item.id; if (item.type === 'messenger' && (item.replyToId || item.replyToMessageId)) { payload.replyToId = item.replyToId || item.replyToMessageId; @@ -355,7 +364,7 @@ export function useOfflineSync(userId?: number): UseOfflineSyncReturn { return; } - _syncLock = (async () => { + _syncLock = (async () => withCrossTabSyncLock(async () => { setStatus('syncing'); try { @@ -564,7 +573,7 @@ export function useOfflineSync(userId?: number): UseOfflineSyncReturn { } finally { _syncLock = null; } - })(); + }))(); await _syncLock; }, []); diff --git a/migrations/0087_conv_message_idempotency.sql b/migrations/0087_conv_message_idempotency.sql new file mode 100644 index 0000000..fe98be4 --- /dev/null +++ b/migrations/0087_conv_message_idempotency.sql @@ -0,0 +1,10 @@ +-- Идемпотентность сообщений мессенджера (conversation_messages): клиентский ключ. +-- Та же защита, что у task_messages (миграция 0086): multi-tab replay +-- офлайн-очереди, ретраи после сетевых сбоев, двойные клики. +-- Partial unique index: NULL-значения (старые сообщения, боты) не конфликтуют. + +ALTER TABLE conversation_messages ADD COLUMN IF NOT EXISTS client_message_id varchar(64); + +CREATE UNIQUE INDEX IF NOT EXISTS conv_messages_client_message_id_unique + ON conversation_messages (client_message_id) + WHERE client_message_id IS NOT NULL; diff --git a/server/routes/messenger.messages.routes.ts b/server/routes/messenger.messages.routes.ts index 475f28c..6953db1 100644 --- a/server/routes/messenger.messages.routes.ts +++ b/server/routes/messenger.messages.routes.ts @@ -139,6 +139,7 @@ export function registerMessengerMessageRoutes(router: Router): void { size: z.number().default(0), mimeType: z.string().optional(), })).optional(), + clientMessageId: z.string().max(64).optional(), }); router.post("/api/messenger/conversations/:id/messages", @@ -171,17 +172,60 @@ export function registerMessengerMessageRoutes(router: Router): void { const orgId = conv?.organizationId ?? req.organizationId!; - const [newMsg] = await db - .insert(conversationMessages) - .values({ - conversationId: convId, - authorId: userId, - message: message.trim(), - replyToId: replyToId ?? null, - mentionedUserIds: mentionedUserIds?.length ? mentionedUserIds : null, - attachments: attachments?.length ? attachments : null, - }) - .returning(); + // Идемпотентность: повтор с тем же clientMessageId — вернуть уже созданное + // сообщение БЕЗ повторного insert и без side-эффектов (SSE/уведомления + // уже отработали при первой записи). Защита от дублей при multi-tab replay + // офлайн-очереди (миграция 0087). + const clientMessageId = parsed.data.clientMessageId ?? null; + if (clientMessageId) { + const [existing] = await db + .select({ id: conversationMessages.id }) + .from(conversationMessages) + .where(and( + eq(conversationMessages.conversationId, convId), + eq(conversationMessages.clientMessageId, clientMessageId) + )) + .limit(1); + if (existing) { + const enrichedExisting = await enrichMessage(existing.id); + return res.status(200).json({ success: true, message: enrichedExisting, deduplicated: true }); + } + } + + let newMsg; + try { + [newMsg] = await db + .insert(conversationMessages) + .values({ + conversationId: convId, + authorId: userId, + message: message.trim(), + replyToId: replyToId ?? null, + mentionedUserIds: mentionedUserIds?.length ? mentionedUserIds : null, + attachments: attachments?.length ? attachments : null, + clientMessageId, + }) + .returning(); + } catch (err) { + // Гонка: параллельный запрос с тем же clientMessageId уже вставил + // сообщение (unique index conv_messages_client_message_id_unique). + const e = err as { code?: string; cause?: { code?: string } }; + if (clientMessageId && (e?.code === '23505' || e?.cause?.code === '23505')) { + const [existing] = await db + .select({ id: conversationMessages.id }) + .from(conversationMessages) + .where(and( + eq(conversationMessages.conversationId, convId), + eq(conversationMessages.clientMessageId, clientMessageId) + )) + .limit(1); + if (existing) { + const enrichedExisting = await enrichMessage(existing.id); + return res.status(200).json({ success: true, message: enrichedExisting, deduplicated: true }); + } + } + throw err; + } const enriched = await enrichMessage(newMsg.id); const members = await getMembersForSSE(convId, orgId); diff --git a/shared/schema.ts b/shared/schema.ts index bd2fd8a..1d65bc1 100644 --- a/shared/schema.ts +++ b/shared/schema.ts @@ -2975,6 +2975,9 @@ export const conversationMessages = pgTable("conversation_messages", { mentionedUserIds: jsonb("mentioned_user_ids").$type(), attachments: jsonb("attachments").$type>(), botButtons: jsonb("bot_buttons").$type>(), + // Идемпотентность: клиентский ключ сообщения (uuid от клиента / id записи очереди). + // Partial unique index — см. миграцию 0087. + clientMessageId: varchar("client_message_id", { length: 64 }), createdAt: timestamp("created_at").defaultNow(), updatedAt: timestamp("updated_at").defaultNow(), isDeleted: boolean("is_deleted").default(false), diff --git a/tests/offline-queue-multitab-race.test.tsx b/tests/offline-queue-multitab-race.test.tsx index 05b3cea..bee7846 100644 --- a/tests/offline-queue-multitab-race.test.tsx +++ b/tests/offline-queue-multitab-race.test.tsx @@ -87,6 +87,8 @@ describe('multi-tab offline queue replay race', () => { beforeEach(() => { sharedQueue.length = 0; replayedPosts = []; + // По умолчанию Web Locks НЕТ (fallback-путь) — гонка воспроизводится + delete (navigator as unknown as { locks?: unknown }).locks; vi.stubGlobal('fetch', vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { const url = String(input); if ((init?.method ?? 'GET') === 'POST') { @@ -99,7 +101,7 @@ describe('multi-tab offline queue replay race', () => { })); }); - it('две вкладки реплеят одну и ту же запись очереди дважды (дубль сообщения)', async () => { + it('без Web Locks две вкладки реплеят одну запись дважды (исходная гонка)', async () => { const useOfflineSyncA = await importFreshHook(); const useOfflineSyncB = await importFreshHook(); @@ -127,9 +129,57 @@ describe('multi-tab offline queue replay race', () => { }); const messagePosts = replayedPosts.filter((p) => p.url === '/api/tasks/1988/messages'); - console.log(`[race-test] POST /api/tasks/1988/messages выполнен ${messagePosts.length} раз(а) для 1 записи очереди`); - // БАГ: одна запись очереди уходит дважды (по разу на каждую открытую вкладку). - // После фикса (cross-tab lock / атомарное изъятие) ожидается 1. + console.log(`[race-test] без Web Locks: POST выполнен ${messagePosts.length} раз(а) для 1 записи`); + // Исходная гонка (документация бага): без кросс-вкладочной координации + // одна запись уходит по разу на каждую вкладку. expect(messagePosts.length).toBe(2); }); + + it('с Web Locks (кросс-вкладочный мьютекс) запись реплеится ровно один раз', async () => { + // Эмуляция navigator.locks: сериализация запросов по имени (как у браузера) + let lockChain: Promise = Promise.resolve(); + const mockLocks = { + request: async (_name: string, cb: () => Promise) => { + const prev = lockChain; + let release!: () => void; + lockChain = new Promise((r) => { release = r; }); + await prev; + try { + return await cb(); + } finally { + release(); + } + }, + }; + Object.defineProperty(navigator, 'locks', { value: mockLocks, configurable: true }); + + const useOfflineSyncA = await importFreshHook(); + const useOfflineSyncB = await importFreshHook(); + + sharedQueue.push({ + id: 'q1', + method: 'POST', + url: '/api/tasks/1988/messages', + body: JSON.stringify({ message: 'тестовое сообщение' }), + token: null, + orgId: '1', + enqueuedAt: Date.now(), + }); + + const tabA = renderHook(() => useOfflineSyncA(1), { wrapper: makeWrapper() }); + const tabB = renderHook(() => useOfflineSyncB(1), { wrapper: makeWrapper() }); + + await act(async () => { + await Promise.all([ + tabA.result.current.sync(), + tabB.result.current.sync(), + ]); + }); + + const messagePosts = replayedPosts.filter((p) => p.url === '/api/tasks/1988/messages'); + console.log(`[race-test] с Web Locks: POST выполнен ${messagePosts.length} раз(а) для 1 записи`); + // ФИКС: первая вкладка под локом реплеит и удаляет запись, вторая видит + // пустую очередь — ровно один POST. + expect(messagePosts.length).toBe(1); + }); });