From 6bf8361936d3e3d84711c398a9c09619c5d7cc6a Mon Sep 17 00:00:00 2001 From: agent_coder Date: Fri, 10 Jul 2026 05:15:11 +0300 Subject: [PATCH] =?UTF-8?q?perf(comment):=20=D1=83=D0=BD=D0=B5=D1=81=D1=82?= =?UTF-8?q?=D0=B8=20Yjs-=D0=BE=D0=B1=D0=BD=D0=BE=D0=B2=D0=BB=D0=B5=D0=BD?= =?UTF-8?q?=D0=B8=D0=B5=20comment-mark=20=D1=81=20HTTP-=D0=BA=D1=80=D0=B8?= =?UTF-8?q?=D1=82=D0=B8=D1=87=D0=B5=D1=81=D0=BA=D0=BE=D0=B3=D0=BE=20=D0=BF?= =?UTF-8?q?=D1=83=D1=82=D0=B8=20(#399)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit resolve/unresolve/delete comment-mark синхронно дёргали collab-gateway (handleYjsEvent) прямо в HTTP-запросе — сетевой раунд-трип к collab на горячем пути. Теперь: - DB-строка пишется синхронно (источник истины) с общим таймстампом; - сама mark-операция уходит идемпотентным COMMENT_MARK_UPDATE-джобом на GENERAL_QUEUE, воркер проигрывает тот же handleYjsEvent; - resolve/unresolve — fire-and-forget (best-effort), delete — await энкью ДО необратимого hard-delete (durability split); - race-guard: устаревшее ПРОТИВОПОЛОЖНОЕ событие (ts <= updatedAt строки и состояние расходится) пропускается, а не флипает mark в устаревшее; - DI-цикл обойдён ленивым moduleRef.get(CollaborationGateway, strict:false). Внутренний цикл: 2 прохода. Правки по внутреннему ревью: `<` → `<=` в race-guard (безопасная обработка суб-миллисекундной ничьи двух противоположных тогглов); задокументирован сознательный компромисс — транзиентный page.updated-broadcast из воркера несёт только {id}, теряя name/avatarUrl «кто редактировал» (lastUpdatedById выставляется верно, косметика, самочинится на следующем реальном редактировании). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../comment.service.apply-suggestion.spec.ts | 48 ++++-- .../comment/comment.service.behavior.spec.ts | 6 +- ...comment.service.dismiss-suggestion.spec.ts | 106 +++++++----- .../comment.service.mark-async.spec.ts | 153 ++++++++++++++++++ .../src/core/comment/comment.service.ts | 92 ++++++++--- .../queue/constants/queue.constants.ts | 3 + .../queue/constants/queue.interface.ts | 27 ++++ ...neral-queue.processor.comment-mark.spec.ts | 151 +++++++++++++++++ .../processors/general-queue.processor.ts | 99 +++++++++++- 9 files changed, 605 insertions(+), 80 deletions(-) create mode 100644 apps/server/src/core/comment/comment.service.mark-async.spec.ts create mode 100644 apps/server/src/integrations/queue/processors/general-queue.processor.comment-mark.spec.ts diff --git a/apps/server/src/core/comment/comment.service.apply-suggestion.spec.ts b/apps/server/src/core/comment/comment.service.apply-suggestion.spec.ts index 4b856962..486f89bb 100644 --- a/apps/server/src/core/comment/comment.service.apply-suggestion.spec.ts +++ b/apps/server/src/core/comment/comment.service.apply-suggestion.spec.ts @@ -5,6 +5,16 @@ import { } from '@nestjs/common'; import { CommentService } from './comment.service'; import { AuditEvent, AuditResource } from '../../common/events/audit-events'; +import { QueueJob } from '../../integrations/queue/constants'; + +// #399: the resolve/unresolve flip and the ephemeral anchor removal are enqueued +// as COMMENT_MARK_UPDATE jobs (off the HTTP path), NOT awaited against the collab +// gateway. applyCommentSuggestion (the document TEXT edit) is untouched — it +// still runs synchronously via the gateway. +const markJob = (generalQueue: any, action: string) => + generalQueue.add.mock.calls.find( + (c: any[]) => c[0] === QueueJob.COMMENT_MARK_UPDATE && c[1]?.action === action, + ); /** * Focused coverage for CommentService.applySuggestion (comment.service.ts). @@ -59,6 +69,7 @@ describe('CommentService — applySuggestion', () => { commentRepo, wsService, collaborationGateway, + generalQueue, auditService, }; } @@ -86,9 +97,15 @@ describe('CommentService — applySuggestion', () => { // --- no replies → ephemeral delete branch ------------------------------- - it('applied=true, no replies → replaces text, hard-deletes, strips the anchor mark, audits APPLIED, outcome=deleted', async () => { - const { service, commentRepo, wsService, collaborationGateway, auditService } = - makeService({ applied: true, currentText: 'new text' }); + it('applied=true, no replies → replaces text, hard-deletes, enqueues the anchor-mark removal, audits APPLIED, outcome=deleted', async () => { + const { + service, + commentRepo, + wsService, + collaborationGateway, + generalQueue, + auditService, + } = makeService({ applied: true, currentText: 'new text' }); const result = await service.applySuggestion(suggestionComment(), user()); @@ -105,12 +122,20 @@ describe('CommentService — applySuggestion', () => { ); // Ephemeral: the redundant comment is hard-deleted (atomic-conditional) and - // its inline anchor mark removed via the deleteCommentMark collab event. + // its inline anchor mark removal is ENQUEUED (#399), no longer a sync gateway + // call. The gateway was only touched for the applyCommentSuggestion text edit. expect(commentRepo.deleteCommentIfChildless).toHaveBeenCalledWith('c-1'); - expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith( + const del = markJob(generalQueue, 'delete'); + expect(del).toBeDefined(); + expect(del[1]).toMatchObject({ + documentName: 'page.page-1', + commentId: 'c-1', + userId: 'user-1', + }); + expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalledWith( 'deleteCommentMark', - 'page.page-1', - expect.objectContaining({ commentId: 'c-1', user: expect.any(Object) }), + expect.anything(), + expect.anything(), ); // No applied stamps are written for a row about to be deleted. expect(appliedPatch(commentRepo)).toBeUndefined(); @@ -258,7 +283,7 @@ describe('CommentService — applySuggestion', () => { // The suggested text is already applied to the document, but between the // hasChildren read and the atomic delete a reply landed. The parent must NOT // be hard-deleted (cascade would destroy the reply); resolve the thread. - const { service, commentRepo, wsService, collaborationGateway } = + const { service, commentRepo, wsService, generalQueue } = makeService({ applied: true, currentText: 'new text' }, false, 0); const result = await service.applySuggestion(suggestionComment(), user()); @@ -275,11 +300,8 @@ describe('CommentService — applySuggestion', () => { .map((c: any[]) => c[0]) .find((p: any) => 'resolvedAt' in p); expect(resolvePatch.resolvedAt).toBeInstanceOf(Date); - expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith( - 'resolveCommentMark', - 'page.page-1', - expect.objectContaining({ commentId: 'c-1', resolved: true }), - ); + // The resolve mark is enqueued (#399), not a sync gateway call. + expect(markJob(generalQueue, 'resolve')).toBeDefined(); expect(result.outcome).toBe('resolved'); }); diff --git a/apps/server/src/core/comment/comment.service.behavior.spec.ts b/apps/server/src/core/comment/comment.service.behavior.spec.ts index 00f2aaaf..8dbe1121 100644 --- a/apps/server/src/core/comment/comment.service.behavior.spec.ts +++ b/apps/server/src/core/comment/comment.service.behavior.spec.ts @@ -313,11 +313,15 @@ describe('CommentService — behavior', () => { }); const [patch] = commentRepo.updateComment.mock.calls[0]; - expect(patch).toEqual({ + // #399: resolve/unresolve now also stamps updatedAt (the async mark + // worker's race-guard reads it to order out-of-order events). The + // resolve-state fields are still cleared to null on unresolve. + expect(patch).toMatchObject({ resolvedAt: null, resolvedById: null, resolvedSource: null, }); + expect(patch.updatedAt).toBeInstanceOf(Date); }); it("notifies the author when SOMEONE ELSE resolves their comment", async () => { diff --git a/apps/server/src/core/comment/comment.service.dismiss-suggestion.spec.ts b/apps/server/src/core/comment/comment.service.dismiss-suggestion.spec.ts index 41fc65be..819071ff 100644 --- a/apps/server/src/core/comment/comment.service.dismiss-suggestion.spec.ts +++ b/apps/server/src/core/comment/comment.service.dismiss-suggestion.spec.ts @@ -1,6 +1,15 @@ import { BadRequestException } from '@nestjs/common'; import { CommentService } from './comment.service'; import { AuditEvent, AuditResource } from '../../common/events/audit-events'; +import { QueueJob } from '../../integrations/queue/constants'; + +// #399: the inline comment-mark op (resolve flip / ephemeral-suggestion anchor +// removal) is now enqueued as a COMMENT_MARK_UPDATE job instead of being awaited +// against the collab gateway on the HTTP path. Find that job by action. +const markJob = (generalQueue: any, action: string) => + generalQueue.add.mock.calls.find( + (c: any[]) => c[0] === QueueJob.COMMENT_MARK_UPDATE && c[1]?.action === action, + ); /** * Coverage for CommentService.dismissSuggestion (#329). Dismiss ("Не применять") @@ -44,7 +53,14 @@ describe('CommentService — dismissSuggestion', () => { auditService, ); - return { service, commentRepo, wsService, collaborationGateway, auditService }; + return { + service, + commentRepo, + wsService, + collaborationGateway, + generalQueue, + auditService, + }; } const suggestionComment = (over?: Partial): any => ({ @@ -62,25 +78,30 @@ describe('CommentService — dismissSuggestion', () => { }); const user = (over?: Partial): any => ({ id: 'user-1', ...over }); - it('no replies → hard-deletes, strips the anchor mark, does NOT touch page text, audits DISMISSED, outcome=deleted', async () => { - const { service, commentRepo, wsService, collaborationGateway, auditService } = - makeService(false); + it('no replies → hard-deletes, enqueues the anchor-mark removal, does NOT touch page text, audits DISMISSED, outcome=deleted', async () => { + const { + service, + commentRepo, + wsService, + collaborationGateway, + generalQueue, + auditService, + } = makeService(false); const result = await service.dismissSuggestion(suggestionComment(), user()); - // Never applies the suggestion to the document. - expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalledWith( - 'applyCommentSuggestion', - expect.anything(), - expect.anything(), - ); - // Hard-delete (atomic-conditional) + strip mark. + // Never applies the suggestion to the document (no sync gateway call at all + // now — the mark op is off the HTTP path, #399). + expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled(); + // Hard-delete (atomic-conditional) + enqueue the anchor-mark strip. expect(commentRepo.deleteCommentIfChildless).toHaveBeenCalledWith('c-1'); - expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith( - 'deleteCommentMark', - 'page.page-1', - expect.objectContaining({ commentId: 'c-1', user: expect.any(Object) }), - ); + const del = markJob(generalQueue, 'delete'); + expect(del).toBeDefined(); + expect(del[1]).toMatchObject({ + documentName: 'page.page-1', + commentId: 'c-1', + userId: 'user-1', + }); expect(wsService.emitCommentEvent).toHaveBeenCalledWith( 'space-1', 'page-1', @@ -96,20 +117,20 @@ describe('CommentService — dismissSuggestion', () => { expect(result.outcome).toBe('deleted'); }); - it('no replies → if the anchor-mark removal FAILS, the row is NOT deleted and the error propagates (#329: no orphan anchor)', async () => { - const { service, commentRepo, wsService, collaborationGateway } = - makeService(false); - // Mark removal is FATAL and runs BEFORE the irreversible row delete: a collab - // failure (e.g. COLLAB_DISABLE_REDIS "no live instance") must abort the whole - // operation, leaving row + mark consistent — never a deleted row with an - // orphan anchor left in the document reporting success. - collaborationGateway.handleYjsEvent = jest.fn(async () => { - throw new Error('requires a live collaboration instance'); + it('no replies → if the anchor-mark ENQUEUE FAILS, the row is NOT deleted and the error propagates (#329/#399: no orphan anchor)', async () => { + const { service, commentRepo, wsService, generalQueue } = makeService(false); + // #399: the mark removal now runs async in a worker, but the ENQUEUE is + // awaited BEFORE the irreversible row delete — so the anchor-removal job is + // durably scheduled before the row can vanish. If even the enqueue fails + // (e.g. Redis down), the whole operation aborts, leaving row + mark + // consistent — never a deleted row with an orphan anchor reporting success. + generalQueue.add = jest.fn(async () => { + throw new Error('queue add failed: no redis'); }); await expect( service.dismissSuggestion(suggestionComment(), user()), - ).rejects.toThrow(/live collaboration/); + ).rejects.toThrow(/queue add failed/); expect(commentRepo.deleteCommentIfChildless).not.toHaveBeenCalled(); expect(wsService.emitCommentEvent).not.toHaveBeenCalledWith( @@ -120,23 +141,29 @@ describe('CommentService — dismissSuggestion', () => { }); it('WITH replies → resolves (not delete), does NOT apply, audits DISMISSED, outcome=resolved', async () => { - const { service, commentRepo, wsService, collaborationGateway, auditService } = - makeService(true); + const { + service, + commentRepo, + collaborationGateway, + generalQueue, + auditService, + } = makeService(true); const result = await service.dismissSuggestion(suggestionComment(), user()); - // Resolved via resolveComment (resolve patch + resolve mark), NOT deleted. + // Resolved via resolveComment (resolve patch + enqueued resolve mark), NOT + // deleted. const resolvePatch = commentRepo.updateComment.mock.calls .map((c: any[]) => c[0]) .find((p: any) => 'resolvedAt' in p); expect(resolvePatch.resolvedAt).toBeInstanceOf(Date); expect(resolvePatch.resolvedById).toBe('user-1'); expect(commentRepo.deleteComment).not.toHaveBeenCalled(); - expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith( - 'resolveCommentMark', - 'page.page-1', - expect.objectContaining({ commentId: 'c-1', resolved: true }), - ); + // No sync gateway call; the resolve mark is enqueued (#399). + expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled(); + const res = markJob(generalQueue, 'resolve'); + expect(res).toBeDefined(); + expect(res[1]).toMatchObject({ documentName: 'page.page-1', commentId: 'c-1' }); // No applied stamp — dismiss does not apply the edit. const appliedPatch = commentRepo.updateComment.mock.calls .map((c: any[]) => c[0]) @@ -156,8 +183,7 @@ describe('CommentService — dismissSuggestion', () => { // but the atomic delete matches 0 rows because a reply landed in the window // between that read and the delete. The parent must NOT be hard-deleted // (a cascade would destroy the just-added reply); the thread is resolved. - const { service, commentRepo, wsService, collaborationGateway } = - makeService(false, 0); + const { service, commentRepo, wsService, generalQueue } = makeService(false, 0); const result = await service.dismissSuggestion(suggestionComment(), user()); @@ -175,11 +201,9 @@ describe('CommentService — dismissSuggestion', () => { .find((p: any) => 'resolvedAt' in p); expect(resolvePatch.resolvedAt).toBeInstanceOf(Date); expect(resolvePatch.resolvedById).toBe('user-1'); - expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith( - 'resolveCommentMark', - 'page.page-1', - expect.objectContaining({ commentId: 'c-1', resolved: true }), - ); + // A resolve mark job is enqueued (the anchor was already delete-marked; the + // resolve mirror is idempotent — #399). + expect(markJob(generalQueue, 'resolve')).toBeDefined(); expect(result.outcome).toBe('resolved'); }); diff --git a/apps/server/src/core/comment/comment.service.mark-async.spec.ts b/apps/server/src/core/comment/comment.service.mark-async.spec.ts new file mode 100644 index 00000000..89a87d1d --- /dev/null +++ b/apps/server/src/core/comment/comment.service.mark-async.spec.ts @@ -0,0 +1,153 @@ +import { CommentService } from './comment.service'; +import { QueueJob } from '../../integrations/queue/constants'; + +/** + * #399: the comment inline-mark update is moved OFF the HTTP critical path. + * resolveComment / unresolve / the ephemeral-suggestion delete must NO LONGER + * await CollaborationGateway.handleYjsEvent (which loaded the whole Y.Doc and + * ran the store pipeline synchronously, ~4.5s p95). Instead they enqueue an + * idempotent COMMENT_MARK_UPDATE job onto the GENERAL_QUEUE with the payload the + * worker replays. + * + * The service is constructed directly with jest mocks (the @InjectQueue tokens + * cannot be resolved by Test.createTestingModule — see comment.service.spec.ts). + */ +describe('CommentService — async comment mark (#399)', () => { + function makeService() { + const commentRepo: any = { + findById: jest.fn(async (id: string) => ({ + id, + content: {}, + spaceId: 'space-1', + pageId: 'page-1', + })), + updateComment: jest.fn(async () => undefined), + hasChildren: jest.fn(async () => false), + deleteCommentIfChildless: jest.fn(async () => 1), + }; + const pageRepo: any = {}; + const wsService: any = { emitCommentEvent: jest.fn() }; + // The gateway MUST NOT be touched on the HTTP path anymore. + const collaborationGateway: any = { + handleYjsEvent: jest.fn(async () => undefined), + }; + const generalQueue: any = { add: jest.fn(() => Promise.resolve()) }; + const notificationQueue: any = { add: jest.fn(async () => undefined) }; + const auditService: any = { log: jest.fn() }; + + const service = new CommentService( + commentRepo, + pageRepo, + wsService, + collaborationGateway, + generalQueue, + notificationQueue, + auditService, + ); + return { + service, + commentRepo, + collaborationGateway, + generalQueue, + auditService, + }; + } + + const comment = (over?: Partial): any => ({ + id: 'c-1', + creatorId: 'user-1', + pageId: 'page-1', + spaceId: 'space-1', + workspaceId: 'ws-1', + ...over, + }); + const user = (over?: Partial): any => ({ id: 'user-1', ...over }); + + const markJob = (generalQueue: any) => + generalQueue.add.mock.calls.find( + (c: any[]) => c[0] === QueueJob.COMMENT_MARK_UPDATE, + ); + + it('resolveComment does NOT call the gateway synchronously, and enqueues a resolve mark job', async () => { + const { service, collaborationGateway, generalQueue } = makeService(); + + await service.resolveComment(comment(), true, user()); + + // The whole point of #399: the Y.Doc mark op is off the HTTP path. + expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled(); + + const job = markJob(generalQueue); + expect(job).toBeDefined(); + expect(job[0]).toBe(QueueJob.COMMENT_MARK_UPDATE); + expect(job[1]).toMatchObject({ + documentName: 'page.page-1', + commentId: 'c-1', + action: 'resolve', + userId: 'user-1', + }); + expect(typeof job[1].ts).toBe('number'); + // ts equals the resolvedAt stamp written to the row (shared timestamp). + const [patch] = (service as any).commentRepo.updateComment.mock.calls[0]; + expect(job[1].ts).toBe((patch.resolvedAt as Date).getTime()); + expect(job[1].ts).toBe((patch.updatedAt as Date).getTime()); + }); + + it('unresolve enqueues an unresolve mark job (action mapped from resolved=false)', async () => { + const { service, collaborationGateway, generalQueue } = makeService(); + + await service.resolveComment(comment(), false, user()); + + expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled(); + const job = markJob(generalQueue); + expect(job[1]).toMatchObject({ + documentName: 'page.page-1', + commentId: 'c-1', + action: 'unresolve', + userId: 'user-1', + }); + }); + + it('dismissing a childless ephemeral suggestion enqueues a delete mark job (not a sync gateway call)', async () => { + const { service, collaborationGateway, generalQueue } = makeService(); + + await service.dismissSuggestion( + comment({ suggestedText: 'new text', selection: 'old', resolvedAt: null }), + user(), + ); + + // The anchor removal is queued, not awaited against the gateway. + expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled(); + const job = markJob(generalQueue); + expect(job).toBeDefined(); + expect(job[1]).toMatchObject({ + documentName: 'page.page-1', + commentId: 'c-1', + action: 'delete', + userId: 'user-1', + }); + expect(typeof job[1].ts).toBe('number'); + }); + + it('awaits the delete ENQUEUE before the irreversible row hard-delete (ordering preserved)', async () => { + const { service, generalQueue, commentRepo } = makeService(); + const order: string[] = []; + generalQueue.add.mockImplementation(async (name: string) => { + order.push(`enqueue:${name}`); + }); + commentRepo.deleteCommentIfChildless.mockImplementation(async () => { + order.push('delete-row'); + return 1; + }); + + await service.dismissSuggestion( + comment({ suggestedText: 'new text', selection: 'old', resolvedAt: null }), + user(), + ); + + // The mark-removal job must be durably queued BEFORE the row disappears. + expect(order).toEqual([ + `enqueue:${QueueJob.COMMENT_MARK_UPDATE}`, + 'delete-row', + ]); + }); +}); diff --git a/apps/server/src/core/comment/comment.service.ts b/apps/server/src/core/comment/comment.service.ts index 8e3d435b..ef97c108 100644 --- a/apps/server/src/core/comment/comment.service.ts +++ b/apps/server/src/core/comment/comment.service.ts @@ -21,6 +21,7 @@ import { CursorPaginationResult } from '@docmost/db/pagination/cursor-pagination import { QueueJob, QueueName } from '../../integrations/queue/constants'; import { extractUserMentionIdsFromJson } from '../../common/helpers/prosemirror/utils'; import { + ICommentMarkUpdateJob, ICommentNotificationJob, ICommentResolvedNotificationJob, } from '../../integrations/queue/constants/queue.interface'; @@ -298,7 +299,11 @@ export class CommentService { // source is cleared alongside resolvedAt/resolvedById. provenance?: AuthProvenanceData, ): Promise { - const resolvedAt = resolved ? new Date() : null; + // One shared timestamp: it stamps resolvedAt AND updatedAt on the row and is + // carried as the mark job's `ts`, so the worker's race-guard can order this + // event against the row's authoritative resolve-state mutation time (#399). + const now = new Date(); + const resolvedAt = resolved ? now : null; const resolvedById = resolved ? authUser.id : null; const isAgent = provenance?.actor === 'agent'; // Set the agent marker only when resolving; on unresolve clear it back to @@ -307,25 +312,33 @@ export class CommentService { const resolvedSource = resolved && isAgent ? 'agent' : null; await this.commentRepo.updateComment( - { resolvedAt, resolvedById, resolvedSource }, + // Bump updatedAt (not editedAt — that drives the "edited" badge) so the + // row records WHEN the resolve state last changed; the async mark worker + // compares its job ts against this to skip a superseded out-of-order event. + { resolvedAt, resolvedById, resolvedSource, updatedAt: now }, comment.id, ); - // Reflect the resolved state on the inline comment mark in the - // collaborative document so all connected clients stay in sync. + // #399: mirror the resolved state onto the inline comment mark OFF the HTTP + // critical path. The DB row above is the source of truth (updated in ms); the + // mark is an eventual mirror for connected clients, and its failure was + // ALREADY swallowed (best-effort warn) — so instead of awaiting the whole + // Y.Doc load + immediate store pipeline (~4.5s p95), enqueue an idempotent, + // retryable COMMENT_MARK_UPDATE job. (Store-pipeline cost itself is #348's + // scope, not duplicated here.) const documentName = `page.${comment.pageId}`; - try { - await this.collaborationGateway.handleYjsEvent( - 'resolveCommentMark', - documentName, - { commentId: comment.id, resolved, user: authUser }, - ); - } catch (error) { + void this.enqueueCommentMarkUpdate( + documentName, + comment.id, + resolved ? 'resolve' : 'unresolve', + now.getTime(), + authUser.id, + ).catch((error) => this.logger.warn( - `Failed to update comment mark for comment ${comment.id}`, + `Failed to enqueue comment mark update for comment ${comment.id}`, error, - ); - } + ), + ); // Notify the comment author when someone else resolves their comment. if (resolved && comment.creatorId !== authUser.id) { @@ -671,23 +684,54 @@ export class CommentService { } /** - * Remove the inline `comment` mark for a comment from the collaborative - * document. FATAL, NOT best-effort: unlike resolveComment (which keeps the row, - * so a failed mark update is recoverable), this is used before an irreversible - * hard-delete, so the mark removal MUST succeed or throw. Under - * COLLAB_DISABLE_REDIS the gateway invokes the deleteCommentMark handler - * directly (never a silent no-op) and a missing live instance surfaces as a - * thrown error, which we let propagate so the caller aborts before deleting. + * Schedule removal of the inline `comment` anchor mark from the collaborative + * document (ephemeral suggestion #329), OFF the HTTP critical path (#399). + * + * ORDERING PRESERVED: we `await` the ENQUEUE (a fast Redis add), not the mark + * op, and the caller only proceeds to the irreversible row hard-delete after + * this resolves. So the anchor-removal job is DURABLY queued before the row + * vanishes — a queue-add failure throws here and aborts the delete (row + mark + * stay consistent), preserving the invariant the old FATAL sync call gave. The + * mark op itself now runs async in the worker: it is idempotent and retried + * (3 attempts), so a transient collab failure self-heals; only an exhausted- + * retries job leaves a DB↔mark divergence, now VISIBLE via BullMQ failed-job + * metrics (was a hard 5xx before). Delete carries no state guard — the row is + * being removed, and stripping an absent mark is a no-op. */ private async deleteCommentMark(comment: Comment, user: User): Promise { const documentName = `page.${comment.pageId}`; - await this.collaborationGateway.handleYjsEvent( - 'deleteCommentMark', + await this.enqueueCommentMarkUpdate( documentName, - { commentId: comment.id, user }, + comment.id, + 'delete', + Date.now(), + user.id, ); } + /** + * Enqueue an idempotent COMMENT_MARK_UPDATE job (#399) — the single path that + * mirrors a comment's inline-mark state into the collab Y.Doc off the HTTP + * response. The worker (GeneralQueueProcessor) runs the SAME handleYjsEvent + * the sync code used, so the mark op is byte-identical. + */ + private enqueueCommentMarkUpdate( + documentName: string, + commentId: string, + action: 'resolve' | 'unresolve' | 'delete', + ts: number, + userId: string, + ): Promise { + const jobData: ICommentMarkUpdateJob = { + documentName, + commentId, + action, + ts, + userId, + }; + return this.generalQueue.add(QueueJob.COMMENT_MARK_UPDATE, jobData); + } + private async queueCommentNotification( content: any, oldMentionIds: string[], diff --git a/apps/server/src/integrations/queue/constants/queue.constants.ts b/apps/server/src/integrations/queue/constants/queue.constants.ts index 0b5765b8..d1ad9b7d 100644 --- a/apps/server/src/integrations/queue/constants/queue.constants.ts +++ b/apps/server/src/integrations/queue/constants/queue.constants.ts @@ -61,6 +61,9 @@ export enum QueueJob { COMMENT_NOTIFICATION = 'comment-notification', COMMENT_RESOLVED_NOTIFICATION = 'comment-resolved-notification', + // #399: off-critical-path mirror of a comment's inline mark into the collab + // Y.Doc (resolve/unresolve flip, or ephemeral-suggestion anchor removal). + COMMENT_MARK_UPDATE = 'comment-mark-update', PAGE_MENTION_NOTIFICATION = 'page-mention-notification', PAGE_PERMISSION_GRANTED = 'page-permission-granted', PAGE_UPDATE_DIGEST = 'page-update-digest', diff --git a/apps/server/src/integrations/queue/constants/queue.interface.ts b/apps/server/src/integrations/queue/constants/queue.interface.ts index 19445944..bcd84889 100644 --- a/apps/server/src/integrations/queue/constants/queue.interface.ts +++ b/apps/server/src/integrations/queue/constants/queue.interface.ts @@ -63,6 +63,33 @@ export interface ICommentNotificationJob { notifyWatchers: boolean; } +/** + * GENERAL_QUEUE payload for the off-critical-path comment inline-mark mirror + * (#399). The comment DB row is the source of truth and is already updated + * synchronously (ms); this job flips/removes the inline `comment` mark in the + * collaborative Y.Doc for connected clients, OFF the HTTP response path, so + * `POST /api/comments/resolve` no longer waits the whole Y.Doc load + store + * pipeline (was ~4.5s p95). The mark op is idempotent, so BullMQ retries are + * safe. + * + * `action`: + * - 'resolve' / 'unresolve' → flip the mark's `resolved` attribute (exactly + * what the synchronous resolveCommentMark path did); + * - 'delete' → strip the anchor mark entirely (ephemeral suggestion #329). + * `ts` is the DB-mutation timestamp (ms). The worker's race-guard uses it (with + * the row's authoritative resolved state) to skip a resolve/unresolve event + * that a newer, opposite event has already superseded (out-of-order drain). + * `userId` supplies the connection-context user the store pipeline attributes + * the change to (persistence.extension reads context.user.id). + */ +export interface ICommentMarkUpdateJob { + documentName: string; + commentId: string; + action: 'resolve' | 'unresolve' | 'delete'; + ts: number; + userId: string; +} + export interface ICommentResolvedNotificationJob { commentId: string; commentCreatorId: string; diff --git a/apps/server/src/integrations/queue/processors/general-queue.processor.comment-mark.spec.ts b/apps/server/src/integrations/queue/processors/general-queue.processor.comment-mark.spec.ts new file mode 100644 index 00000000..7adf30ed --- /dev/null +++ b/apps/server/src/integrations/queue/processors/general-queue.processor.comment-mark.spec.ts @@ -0,0 +1,151 @@ +import { Job } from 'bullmq'; +import { GeneralQueueProcessor } from './general-queue.processor'; +import { QueueJob } from '../constants'; +import { ICommentMarkUpdateJob } from '../constants/queue.interface'; + +/** + * #399: the GENERAL_QUEUE worker replays the comment inline-mark op that used to + * run synchronously on the HTTP path. It must call the SAME gateway handler with + * the SAME semantics (resolve/unresolve → flip the `resolved` attribute; delete + * → strip the anchor), and its timestamp race-guard must skip an event a newer, + * opposite event already superseded. + */ +describe('GeneralQueueProcessor — COMMENT_MARK_UPDATE (#399)', () => { + function makeProc() { + const collaborationGateway: any = { + handleYjsEvent: jest.fn(async () => undefined), + }; + const commentRepo: any = { findById: jest.fn() }; + // #399: the processor resolves CollaborationGateway lazily via ModuleRef + // (strict:false) to avoid a DI cycle; the fake returns our gateway spy. + const moduleRef: any = { get: jest.fn(() => collaborationGateway) }; + const proc = new GeneralQueueProcessor( + {} as any, // db + {} as any, // backlinkRepo + {} as any, // watcherRepo + commentRepo, + moduleRef, + ); + return { proc, collaborationGateway, commentRepo }; + } + + const job = (data: ICommentMarkUpdateJob): Job => + ({ name: QueueJob.COMMENT_MARK_UPDATE, data }) as unknown as Job; + + const base = { + documentName: 'page.page-1', + commentId: 'c-1', + userId: 'user-1', + }; + + it('resolve → resolveCommentMark with resolved:true and the same-shape args', async () => { + const { proc, collaborationGateway, commentRepo } = makeProc(); + const ts = 1000; + // Row reflects the resolve (source of truth), stamped at the same ts. + commentRepo.findById.mockResolvedValue({ + id: 'c-1', + resolvedAt: new Date(ts), + updatedAt: new Date(ts), + }); + + await proc.process(job({ ...base, action: 'resolve', ts })); + + expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledTimes(1); + expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith( + 'resolveCommentMark', + 'page.page-1', + { commentId: 'c-1', resolved: true, user: { id: 'user-1' } }, + ); + }); + + it('unresolve → resolveCommentMark with resolved:false', async () => { + const { proc, collaborationGateway, commentRepo } = makeProc(); + const ts = 2000; + commentRepo.findById.mockResolvedValue({ + id: 'c-1', + resolvedAt: null, + updatedAt: new Date(ts), + }); + + await proc.process(job({ ...base, action: 'unresolve', ts })); + + expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith( + 'resolveCommentMark', + 'page.page-1', + { commentId: 'c-1', resolved: false, user: { id: 'user-1' } }, + ); + }); + + it('delete → deleteCommentMark (strip the anchor), no row lookup / no state guard', async () => { + const { proc, collaborationGateway, commentRepo } = makeProc(); + + await proc.process(job({ ...base, action: 'delete', ts: 123 })); + + expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith( + 'deleteCommentMark', + 'page.page-1', + { commentId: 'c-1', user: { id: 'user-1' } }, + ); + // Delete carries no state guard — the row is (being) removed. + expect(commentRepo.findById).not.toHaveBeenCalled(); + }); + + it('SKIPS a stale resolve superseded by a newer unresolve (row unresolved, job ts older)', async () => { + const { proc, collaborationGateway, commentRepo } = makeProc(); + // A later unresolve already set the row: resolvedAt null, updatedAt = 5000. + commentRepo.findById.mockResolvedValue({ + id: 'c-1', + resolvedAt: null, + updatedAt: new Date(5000), + }); + + // Stale resolve job enqueued at ts=1000 (< 5000), intends resolved=true, + // but the row's authoritative state is unresolved → skip. + await proc.process(job({ ...base, action: 'resolve', ts: 1000 })); + + expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled(); + }); + + it('SKIPS a stale unresolve superseded by a newer resolve', async () => { + const { proc, collaborationGateway, commentRepo } = makeProc(); + commentRepo.findById.mockResolvedValue({ + id: 'c-1', + resolvedAt: new Date(5000), + updatedAt: new Date(5000), + }); + + await proc.process(job({ ...base, action: 'unresolve', ts: 1000 })); + + expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled(); + }); + + it('applies when the row state agrees even if ts is older (idempotent, not a stale flip)', async () => { + const { proc, collaborationGateway, commentRepo } = makeProc(); + // Row is resolved and its updatedAt is newer than the job ts, but the state + // AGREES with the job → this is a harmless idempotent replay, not a stale + // opposite event, so it must still apply. + commentRepo.findById.mockResolvedValue({ + id: 'c-1', + resolvedAt: new Date(9000), + updatedAt: new Date(9000), + }); + + await proc.process(job({ ...base, action: 'resolve', ts: 1000 })); + + expect(collaborationGateway.handleYjsEvent).toHaveBeenCalledWith( + 'resolveCommentMark', + 'page.page-1', + { commentId: 'c-1', resolved: true, user: { id: 'user-1' } }, + ); + }); + + it('skips (no throw) when the comment row has vanished', async () => { + const { proc, collaborationGateway, commentRepo } = makeProc(); + commentRepo.findById.mockResolvedValue(undefined); + + await expect( + proc.process(job({ ...base, action: 'resolve', ts: 1000 })), + ).resolves.toBeUndefined(); + expect(collaborationGateway.handleYjsEvent).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/server/src/integrations/queue/processors/general-queue.processor.ts b/apps/server/src/integrations/queue/processors/general-queue.processor.ts index 617cee10..e75cd8c7 100644 --- a/apps/server/src/integrations/queue/processors/general-queue.processor.ts +++ b/apps/server/src/integrations/queue/processors/general-queue.processor.ts @@ -4,6 +4,7 @@ import { Job } from 'bullmq'; import { QueueJob, QueueName } from '../constants'; import { IAddPageWatchersJob, + ICommentMarkUpdateJob, IPageBacklinkJob, } from '../constants/queue.interface'; import { InjectKysely } from 'nestjs-kysely'; @@ -13,8 +14,11 @@ import { WatcherRepo, WatcherType, } from '@docmost/db/repos/watcher/watcher.repo'; -import { InsertableWatcher } from '@docmost/db/types/entity.types'; +import { InsertableWatcher, User } from '@docmost/db/types/entity.types'; import { processBacklinks } from '../tasks/backlinks.task'; +import { ModuleRef } from '@nestjs/core'; +import { CollaborationGateway } from '../../../collaboration/collaboration.gateway'; +import { CommentRepo } from '@docmost/db/repos/comment/comment.repo'; @Processor(QueueName.GENERAL_QUEUE) export class GeneralQueueProcessor @@ -22,14 +26,32 @@ export class GeneralQueueProcessor implements OnModuleDestroy { private readonly logger = new Logger(GeneralQueueProcessor.name); + // #399: CollaborationGateway lives in CollaborationModule. We resolve it lazily + // via ModuleRef instead of importing that module into the @Global QueueModule — + // CollaborationModule's own HistoryProcessor injects this module's global + // GENERAL_QUEUE token, so a static import edge here would form a DI cycle. A + // lazy strict:false lookup (cached) sidesteps it; the gateway is a singleton in + // both the API-server and collab processes that run this worker. + private collaborationGateway?: CollaborationGateway; constructor( @InjectKysely() private readonly db: KyselyDB, private readonly backlinkRepo: BacklinkRepo, private readonly watcherRepo: WatcherRepo, + private readonly commentRepo: CommentRepo, + private readonly moduleRef: ModuleRef, ) { super(); } + private getCollaborationGateway(): CollaborationGateway { + if (!this.collaborationGateway) { + this.collaborationGateway = this.moduleRef.get(CollaborationGateway, { + strict: false, + }); + } + return this.collaborationGateway; + } + async process(job: Job): Promise { try { switch (job.name) { @@ -56,12 +78,87 @@ export class GeneralQueueProcessor ); break; } + + case QueueJob.COMMENT_MARK_UPDATE: { + await this.processCommentMarkUpdate( + job.data as ICommentMarkUpdateJob, + ); + break; + } } } catch (err) { throw err; } } + /** + * #399: apply a comment's inline-mark mirror in the collab Y.Doc, off the HTTP + * critical path. Runs the SAME gateway path the synchronous comment.service + * code used (byte-identical mark op): + * - resolve / unresolve → resolveCommentMark (flip the `resolved` attribute); + * - delete → deleteCommentMark (strip the ephemeral-suggestion anchor #329). + * The op is idempotent, so a BullMQ retry is safe. Throwing propagates to + * WorkerHost → the job is retried and, on exhaustion, surfaces in failed-job + * metrics (the divergence is now visible rather than a silently-swallowed warn). + */ + private async processCommentMarkUpdate( + data: ICommentMarkUpdateJob, + ): Promise { + const { documentName, commentId, action, ts, userId } = data; + // Minimal connection-context user: the store pipeline reads context.user.id + // to attribute the change (persistence.extension). The mark mutation itself + // does not depend on the user, so the op stays byte-identical. Deliberate + // trade-off: the store pipeline's transient `page.updated` broadcast carries + // only { id } here, so its live "who edited" badge loses name/avatarUrl for + // this async mark replay. lastUpdatedById is still set correctly; the diff is + // cosmetic and self-heals on the next real edit — worth it to stay off the + // HTTP path and avoid re-loading the users row. + const user = { id: userId } as User; + + if (action === 'delete') { + await this.getCollaborationGateway().handleYjsEvent( + 'deleteCommentMark', + documentName, + { commentId, user }, + ); + return; + } + + // resolve / unresolve. The comment row is written SYNCHRONOUSLY before this + // job is enqueued, so it is the source of truth for the final resolved state + // and its updatedAt records when that state last changed. Race-guard: if a + // newer, OPPOSITE event has already superseded this one (its ts is older than + // the row's last resolve-state mutation AND the row's current resolved state + // disagrees with what this job intends — e.g. an unresolve that drained ahead + // of this resolve), skip it rather than flip the mark to a stale state. + const comment = await this.commentRepo.findById(commentId); + if (!comment) { + // The comment vanished (e.g. hard-deleted) → nothing left to mirror. + return; + } + const wantResolved = action === 'resolve'; + const rowResolved = comment.resolvedAt != null; + const rowMutatedAt = new Date(comment.updatedAt).getTime(); + // `<=`, not `<`: on a sub-millisecond tie (two opposite toggles stamped in + // the same ms) skip the disagreeing job rather than let queue order decide. + // The consistent job (whose intent matches the row) short-circuits on the + // first condition, so a real update is never dropped; only a mark that both + // disagrees with the row AND is no newer than it is discarded. + if (rowResolved !== wantResolved && ts <= rowMutatedAt) { + this.logger.debug( + `Skipping stale comment mark '${action}' for ${commentId} ` + + `(job ts ${ts} < row ${rowMutatedAt}, row resolved=${rowResolved})`, + ); + return; + } + + await this.getCollaborationGateway().handleYjsEvent( + 'resolveCommentMark', + documentName, + { commentId, resolved: wantResolved, user }, + ); + } + @OnWorkerEvent('active') onActive(job: Job) { this.logger.debug(`Processing ${job.name} job`);