Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d84e5ddbad | |||
| 6bf8361936 |
@@ -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');
|
||||
});
|
||||
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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>): any => ({
|
||||
@@ -62,25 +78,30 @@ describe('CommentService — dismissSuggestion', () => {
|
||||
});
|
||||
const user = (over?: Partial<any>): 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');
|
||||
});
|
||||
|
||||
|
||||
@@ -0,0 +1,179 @@
|
||||
import { Logger } from '@nestjs/common';
|
||||
import { CommentService } from './comment.service';
|
||||
import { QueueJob } from '../../integrations/queue/constants';
|
||||
|
||||
// Flush pending microtasks so a fire-and-forget `.catch(...)` runs before we assert.
|
||||
const flushMicrotasks = () => new Promise((r) => setImmediate(r));
|
||||
|
||||
/**
|
||||
* #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>): any => ({
|
||||
id: 'c-1',
|
||||
creatorId: 'user-1',
|
||||
pageId: 'page-1',
|
||||
spaceId: 'space-1',
|
||||
workspaceId: 'ws-1',
|
||||
...over,
|
||||
});
|
||||
const user = (over?: Partial<any>): 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',
|
||||
]);
|
||||
});
|
||||
|
||||
it('resolve is fire-and-forget: a queue-add rejection does NOT fail the HTTP call (best-effort warn)', async () => {
|
||||
const { service, generalQueue } = makeService();
|
||||
// The queue is unavailable — the whole point of #399 is that this must NOT
|
||||
// propagate out of resolveComment onto the HTTP request.
|
||||
const queueErr = new Error('queue is down');
|
||||
generalQueue.add.mockRejectedValue(queueErr);
|
||||
const warnSpy = jest
|
||||
.spyOn(Logger.prototype, 'warn')
|
||||
.mockImplementation(() => undefined);
|
||||
|
||||
// Must resolve, never throw, even though the enqueue rejects.
|
||||
await expect(service.resolveComment(comment(), true, user())).resolves.not.toThrow();
|
||||
|
||||
// The rejection is swallowed on a microtask AFTER the method returns; flush it.
|
||||
await flushMicrotasks();
|
||||
expect(warnSpy).toHaveBeenCalledWith(
|
||||
expect.stringContaining('Failed to enqueue comment mark update for comment c-1'),
|
||||
queueErr,
|
||||
);
|
||||
warnSpy.mockRestore();
|
||||
});
|
||||
});
|
||||
@@ -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<Comment> {
|
||||
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<void> {
|
||||
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<unknown> {
|
||||
const jobData: ICommentMarkUpdateJob = {
|
||||
documentName,
|
||||
commentId,
|
||||
action,
|
||||
ts,
|
||||
userId,
|
||||
};
|
||||
return this.generalQueue.add(QueueJob.COMMENT_MARK_UPDATE, jobData);
|
||||
}
|
||||
|
||||
private async queueCommentNotification(
|
||||
content: any,
|
||||
oldMentionIds: string[],
|
||||
|
||||
@@ -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',
|
||||
|
||||
@@ -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;
|
||||
|
||||
+151
@@ -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();
|
||||
});
|
||||
});
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
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`);
|
||||
|
||||
+32
-115
@@ -167,35 +167,6 @@ function isUuid(value: string): boolean {
|
||||
return typeof value === "string" && UUID_RE.test(value);
|
||||
}
|
||||
|
||||
/**
|
||||
* Collab-token cache TTL in milliseconds (issue #435). Read fresh from the
|
||||
* environment on every mint — like collab-session.ts readConfig — so tests and a
|
||||
* live rollback can change it without reloading the module.
|
||||
*
|
||||
* Why a cache at all: the live CollabSession registry (#400/#431) keys sessions
|
||||
* on (wsUrl, pageId, collabToken) for identity isolation (invariant 4). But BOTH
|
||||
* collab-token sources mint a FRESH token per mutation — the in-app provider
|
||||
* re-signs a JWT whose iat/exp (seconds) changes every second, and the external
|
||||
* MCP POSTs /auth/collab-token each call — so the token in the key changed on
|
||||
* every op and the session was almost never reused (connect-storms, 25s
|
||||
* timeouts, zombie sessions). Caching the token per-client keeps the key stable
|
||||
* across a burst of mutations so ONE session is reused.
|
||||
*
|
||||
* Default 5 min: well under the 24h collab-token lifetime AND <= the collab
|
||||
* session max-age (10 min, MCP_COLLAB_SESSION_MAX_AGE_MS), so the
|
||||
* permission-staleness window is not widened beyond what #431 already accepted.
|
||||
* The rollback knob is an EXPLICIT 0 (or a negative number): that DISABLES the
|
||||
* cache — an exact fetch-per-call legacy path, mirroring how idleMs<=0 disables
|
||||
* the session cache. Unset OR unparseable (e.g. a typo like "5min", "abc") falls
|
||||
* back to the 5-min default with the cache ON — parseInt yields NaN, which is
|
||||
* treated as "not configured", not as "disabled". So to turn the cache off you
|
||||
* must set the value to exactly 0, not to garbage.
|
||||
*/
|
||||
function readCollabTokenTtlMs(): number {
|
||||
const raw = parseInt(process.env.MCP_COLLAB_TOKEN_TTL_MS ?? "", 10);
|
||||
return Number.isFinite(raw) ? Math.max(0, raw) : 5 * 60 * 1000;
|
||||
}
|
||||
|
||||
export class DocmostClient {
|
||||
private client: AxiosInstance;
|
||||
private token: string | null = null;
|
||||
@@ -234,15 +205,6 @@ export class DocmostClient {
|
||||
// resolvePageId), so only slugId->uuid entries are stored/read here.
|
||||
private pageIdCache = new Map<string, string>();
|
||||
|
||||
// Collab-token cache (issue #435): the last minted collab token plus the
|
||||
// wall-clock time it was minted, so a burst of content mutations reuses ONE
|
||||
// token and therefore ONE live CollabSession (whose registry key includes the
|
||||
// token — #400 invariant 4). Per-instance: a DocmostClient is built per
|
||||
// user/per chat request, so a cached token can never leak across identities.
|
||||
// Reset whenever the client's identity changes (login() / this.token cleared);
|
||||
// bypassed on a forced refresh (the 401/403 reauth path). null = no token yet.
|
||||
private collabTokenCache: { token: string; mintedAt: number } | null = null;
|
||||
|
||||
// Two construction forms:
|
||||
// - new DocmostClient(config) // discriminated union (current)
|
||||
// - new DocmostClient(baseURL, email, password) // legacy positional creds
|
||||
@@ -311,11 +273,8 @@ export class DocmostClient {
|
||||
|
||||
if (config && isAuthError && !config._retry && !isLoginRequest) {
|
||||
config._retry = true;
|
||||
// Drop the stale token + Authorization header before re-login. Also
|
||||
// clear the collab-token cache (#435): a new identity/login must not
|
||||
// keep serving a collab token minted under the old one.
|
||||
// Drop the stale token + Authorization header before re-login.
|
||||
this.token = null;
|
||||
this.collabTokenCache = null;
|
||||
delete this.client.defaults.headers.common["Authorization"];
|
||||
try {
|
||||
await this.login();
|
||||
@@ -364,9 +323,6 @@ export class DocmostClient {
|
||||
throw new Error("getToken returned an empty token");
|
||||
}
|
||||
this.token = token;
|
||||
// Identity (re)established: drop any collab token minted under a
|
||||
// previous identity so the #435 cache can never outlive it.
|
||||
this.collabTokenCache = null;
|
||||
this.client.defaults.headers.common["Authorization"] =
|
||||
`Bearer ${token}`;
|
||||
})
|
||||
@@ -389,34 +345,8 @@ export class DocmostClient {
|
||||
* by this.client's response interceptor; this helper replicates that
|
||||
* behaviour for collab-token requests: ensure a token, try once, and on an
|
||||
* expired-token auth error perform a fresh login and retry exactly once.
|
||||
*
|
||||
* Collab-token cache (issue #435): both sources — the getCollabToken provider
|
||||
* (in-app agent) AND the REST /auth/collab-token endpoint (external MCP) — mint
|
||||
* a FRESH token per call, whose string therefore changes every op. Since the
|
||||
* live CollabSession registry keys on the token string (#400/#431 invariant 4),
|
||||
* that churned the key and defeated session reuse. So we cache the last minted
|
||||
* token per-client for readCollabTokenTtlMs() and hand it back for a burst of
|
||||
* mutations, keeping the session key stable. `forceRefresh` bypasses the cache
|
||||
* (the 401/403 reauth retry uses it, so the retry cannot be handed the same
|
||||
* stale token that just failed — otherwise reauth would be a no-op). TTL 0
|
||||
* disables the cache: exact fetch-per-call legacy behaviour.
|
||||
*/
|
||||
private async getCollabTokenWithReauth(
|
||||
forceRefresh = false,
|
||||
): Promise<string> {
|
||||
const ttl = readCollabTokenTtlMs();
|
||||
// Serve the cached collab token while it is still fresh (identity isolation
|
||||
// is preserved: the cache is a per-instance field on a client built per
|
||||
// user/per chat request, and it is cleared on every identity change).
|
||||
if (
|
||||
!forceRefresh &&
|
||||
ttl > 0 &&
|
||||
this.collabTokenCache &&
|
||||
Date.now() - this.collabTokenCache.mintedAt < ttl
|
||||
) {
|
||||
return this.collabTokenCache.token;
|
||||
}
|
||||
|
||||
private async getCollabTokenWithReauth(): Promise<string> {
|
||||
// Collab-token PROVIDER path: when a getCollabToken provider was supplied
|
||||
// (the internal agent's provenance collab token), use it instead of the
|
||||
// REST /auth/collab-token endpoint. Re-invoke it once on a 401/403 (e.g. the
|
||||
@@ -427,13 +357,23 @@ export class DocmostClient {
|
||||
if (typeof token !== "string" || token.length === 0) {
|
||||
throw new Error("getCollabToken returned an empty token");
|
||||
}
|
||||
return this.rememberCollabToken(token, ttl);
|
||||
return token;
|
||||
} catch (e) {
|
||||
// On an auth error retry EXACTLY once, forcing a refresh so the retry
|
||||
// re-invokes the provider (bypassing the cache) for a genuinely fresh
|
||||
// token. `!forceRefresh` bounds it to a single retry (no loop).
|
||||
if (this.isCollabAuthError(e) && !forceRefresh) {
|
||||
return this.getCollabTokenWithReauth(true);
|
||||
const axiosStatus = axios.isAxiosError(e)
|
||||
? e.response?.status
|
||||
: undefined;
|
||||
const attachedStatus = (e as any)?.status;
|
||||
const isAuthError =
|
||||
axiosStatus === 401 ||
|
||||
axiosStatus === 403 ||
|
||||
attachedStatus === 401 ||
|
||||
attachedStatus === 403;
|
||||
if (isAuthError) {
|
||||
const token = await this.getCollabTokenFn();
|
||||
if (typeof token !== "string" || token.length === 0) {
|
||||
throw new Error("getCollabToken returned an empty token");
|
||||
}
|
||||
return token;
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
@@ -441,51 +381,28 @@ export class DocmostClient {
|
||||
|
||||
await this.ensureAuthenticated();
|
||||
try {
|
||||
const token = await getCollabToken(this.apiUrl, this.token!);
|
||||
return this.rememberCollabToken(token, ttl);
|
||||
return await getCollabToken(this.apiUrl, this.token!);
|
||||
} catch (e) {
|
||||
// getCollabToken wraps the AxiosError in a plain Error but attaches the
|
||||
// HTTP status as `.status`, so isCollabAuthError detects an auth failure
|
||||
// via either the raw AxiosError shape OR the attached status.
|
||||
if (this.isCollabAuthError(e) && !forceRefresh) {
|
||||
// Fresh login (which clears this.token AND the collab-token cache), then
|
||||
// retry exactly once with the cache bypassed via forceRefresh.
|
||||
// HTTP status as `.status`, so detect an auth failure via either the raw
|
||||
// AxiosError shape OR the attached status.
|
||||
const axiosStatus = axios.isAxiosError(e)
|
||||
? e.response?.status
|
||||
: undefined;
|
||||
const attachedStatus = (e as any)?.status;
|
||||
const isAuthError =
|
||||
axiosStatus === 401 ||
|
||||
axiosStatus === 403 ||
|
||||
attachedStatus === 401 ||
|
||||
attachedStatus === 403;
|
||||
if (isAuthError) {
|
||||
await this.login();
|
||||
return this.getCollabTokenWithReauth(true);
|
||||
return await getCollabToken(this.apiUrl, this.token!);
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Store a freshly minted collab token in the per-client cache (issue #435) and
|
||||
* return it unchanged. No-op write when the cache is disabled (ttl<=0) or the
|
||||
* token is empty, so a disabled cache is exact fetch-per-call legacy behaviour
|
||||
* and a bad token is never cached.
|
||||
*/
|
||||
private rememberCollabToken(token: string, ttl: number): string {
|
||||
if (ttl > 0 && typeof token === "string" && token.length > 0) {
|
||||
this.collabTokenCache = { token, mintedAt: Date.now() };
|
||||
}
|
||||
return token;
|
||||
}
|
||||
|
||||
/**
|
||||
* True when an error carries a 401/403 — either as a raw AxiosError
|
||||
* (`error.response.status`) or as the plain-Error `.status` that
|
||||
* lib/auth-utils.getCollabToken attaches after wrapping the AxiosError.
|
||||
*/
|
||||
private isCollabAuthError(e: unknown): boolean {
|
||||
const axiosStatus = axios.isAxiosError(e) ? e.response?.status : undefined;
|
||||
const attachedStatus = (e as any)?.status;
|
||||
return (
|
||||
axiosStatus === 401 ||
|
||||
axiosStatus === 403 ||
|
||||
attachedStatus === 401 ||
|
||||
attachedStatus === 403
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Connect to the collaboration websocket, read the live doc, apply
|
||||
* `transform`, write the result, and wait for the server to persist it —
|
||||
|
||||
@@ -1,282 +0,0 @@
|
||||
// Unit tests for the collab-token cache (issue #435). The live CollabSession
|
||||
// registry (#400/#431) keys sessions on (wsUrl, pageId, collabToken), so a token
|
||||
// string that changes every op defeats reuse. This cache holds the last minted
|
||||
// token per DocmostClient for MCP_COLLAB_TOKEN_TTL_MS so a burst of mutations
|
||||
// reuses ONE token -> ONE session. These tests exercise both mint sources:
|
||||
// - the getCollabToken PROVIDER path (in-app agent), via a counting provider fn;
|
||||
// - the REST /auth/collab-token path (external MCP), via a mock http server.
|
||||
// getCollabTokenWithReauth is private in TS but a plain method on the compiled
|
||||
// build, so the tests call it directly (same convention as reauth.test.mjs).
|
||||
import { test, afterEach, after } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import http from "node:http";
|
||||
import { DocmostClient } from "../../build/client.js";
|
||||
|
||||
// Restore the env knob after each test so cases do not leak into one another.
|
||||
const ENV_KEY = "MCP_COLLAB_TOKEN_TTL_MS";
|
||||
afterEach(() => {
|
||||
delete process.env[ENV_KEY];
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Small mock server for the REST /auth/collab-token path. Counts collab-token
|
||||
// mints and can be told to 401 the first N of them (to drive the reauth retry).
|
||||
// ---------------------------------------------------------------------------
|
||||
function readBody(req) {
|
||||
return new Promise((resolve) => {
|
||||
let raw = "";
|
||||
req.on("data", (c) => (raw += c));
|
||||
req.on("end", () => resolve(raw));
|
||||
});
|
||||
}
|
||||
function sendJson(res, status, obj, extra = {}) {
|
||||
res.writeHead(status, { "Content-Type": "application/json", ...extra });
|
||||
res.end(JSON.stringify(obj));
|
||||
}
|
||||
|
||||
const openServers = [];
|
||||
after(async () => {
|
||||
await Promise.all(
|
||||
openServers.map((s) => new Promise((r) => s.close(r))),
|
||||
);
|
||||
});
|
||||
|
||||
// state: { collabCalls, loginCalls, unauthorizedCollabHits }
|
||||
function spawnCollabServer(state, { collabAuthFailsFor = 0 } = {}) {
|
||||
return new Promise((resolve) => {
|
||||
const server = http.createServer(async (req, res) => {
|
||||
await readBody(req);
|
||||
if (req.url === "/api/auth/login") {
|
||||
state.loginCalls++;
|
||||
// A fresh authToken per login so an identity change is observable.
|
||||
sendJson(res, 200, { success: true }, {
|
||||
"Set-Cookie": `authToken=login-${state.loginCalls}; Path=/; HttpOnly`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (req.url === "/api/auth/collab-token") {
|
||||
state.collabCalls++;
|
||||
if (state.collabCalls <= collabAuthFailsFor) {
|
||||
sendJson(res, 401, { message: "Unauthorized" });
|
||||
return;
|
||||
}
|
||||
// Unique token per mint so a stale cached value is distinguishable.
|
||||
sendJson(res, 200, { data: { token: `collab-${state.collabCalls}` } });
|
||||
return;
|
||||
}
|
||||
sendJson(res, 404, { message: "not found" });
|
||||
});
|
||||
server.listen(0, "127.0.0.1", () => {
|
||||
openServers.push(server);
|
||||
resolve(`http://127.0.0.1:${server.address().port}/api`);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
// PROVIDER path (in-app agent getCollabToken fn)
|
||||
// ===========================================================================
|
||||
|
||||
// A counting provider that returns a distinct token each call so a cached
|
||||
// (reused) token is visibly the SAME string while a fresh mint is different.
|
||||
function countingProvider() {
|
||||
let n = 0;
|
||||
const fn = async () => {
|
||||
n++;
|
||||
return `provider-token-${n}`;
|
||||
};
|
||||
return {
|
||||
fn,
|
||||
get calls() {
|
||||
return n;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
test("within TTL, repeated calls return the SAME token and mint ONCE (provider path)", async () => {
|
||||
process.env[ENV_KEY] = "300000"; // 5 min
|
||||
const p = countingProvider();
|
||||
const client = new DocmostClient({
|
||||
apiUrl: "http://127.0.0.1:1/api",
|
||||
getToken: async () => "access",
|
||||
getCollabToken: p.fn,
|
||||
});
|
||||
|
||||
const a = await client.getCollabTokenWithReauth();
|
||||
const b = await client.getCollabTokenWithReauth();
|
||||
const c = await client.getCollabTokenWithReauth();
|
||||
|
||||
assert.equal(a, "provider-token-1");
|
||||
assert.equal(b, a, "second call reuses the cached token");
|
||||
assert.equal(c, a, "third call reuses the cached token");
|
||||
assert.equal(p.calls, 1, "the provider is invoked exactly once within the TTL");
|
||||
});
|
||||
|
||||
test("after TTL expiry a new token is minted (provider path)", async () => {
|
||||
process.env[ENV_KEY] = "20"; // 20ms TTL
|
||||
const p = countingProvider();
|
||||
const client = new DocmostClient({
|
||||
apiUrl: "http://127.0.0.1:1/api",
|
||||
getToken: async () => "access",
|
||||
getCollabToken: p.fn,
|
||||
});
|
||||
|
||||
const a = await client.getCollabTokenWithReauth();
|
||||
await new Promise((r) => setTimeout(r, 40)); // let the TTL lapse
|
||||
const b = await client.getCollabTokenWithReauth();
|
||||
|
||||
assert.equal(a, "provider-token-1");
|
||||
assert.equal(b, "provider-token-2", "a fresh token is minted after expiry");
|
||||
assert.equal(p.calls, 2);
|
||||
});
|
||||
|
||||
test("MCP_COLLAB_TOKEN_TTL_MS=0 disables the cache: mint on EVERY call (provider path)", async () => {
|
||||
process.env[ENV_KEY] = "0";
|
||||
const p = countingProvider();
|
||||
const client = new DocmostClient({
|
||||
apiUrl: "http://127.0.0.1:1/api",
|
||||
getToken: async () => "access",
|
||||
getCollabToken: p.fn,
|
||||
});
|
||||
|
||||
await client.getCollabTokenWithReauth();
|
||||
await client.getCollabTokenWithReauth();
|
||||
await client.getCollabTokenWithReauth();
|
||||
|
||||
assert.equal(p.calls, 3, "cache disabled -> exact fetch-per-call legacy path");
|
||||
});
|
||||
|
||||
test("a 401 triggers the internal reauth retry, which bypasses the cache and mints fresh (provider path)", async () => {
|
||||
process.env[ENV_KEY] = "300000";
|
||||
let n = 0;
|
||||
const provider = async () => {
|
||||
n++;
|
||||
if (n === 1) {
|
||||
// The FIRST mint fails with an auth error; the internal reauth retry must
|
||||
// re-invoke the provider (bypassing the empty cache) for a fresh token.
|
||||
const err = new Error("collab token expired");
|
||||
err.status = 401;
|
||||
throw err;
|
||||
}
|
||||
return `provider-token-${n}`;
|
||||
};
|
||||
const client = new DocmostClient({
|
||||
apiUrl: "http://127.0.0.1:1/api",
|
||||
getToken: async () => "access",
|
||||
getCollabToken: provider,
|
||||
});
|
||||
|
||||
// Cache is empty: mint #1 401s -> the reauth retry mints #2 and caches it.
|
||||
const tok = await client.getCollabTokenWithReauth();
|
||||
assert.equal(tok, "provider-token-2", "the post-401 retry token wins");
|
||||
assert.equal(n, 2, "exactly one failed mint + one retry, no loop");
|
||||
|
||||
// The retried token is what got cached (no extra mint on a cache hit).
|
||||
const cached = await client.getCollabTokenWithReauth();
|
||||
assert.equal(cached, "provider-token-2");
|
||||
assert.equal(n, 2, "served from cache, provider not re-invoked");
|
||||
});
|
||||
|
||||
test("forceRefresh=true bypasses a warm cache and mints a fresh token (provider path)", async () => {
|
||||
process.env[ENV_KEY] = "300000";
|
||||
const p = countingProvider();
|
||||
const client = new DocmostClient({
|
||||
apiUrl: "http://127.0.0.1:1/api",
|
||||
getToken: async () => "access",
|
||||
getCollabToken: p.fn,
|
||||
});
|
||||
|
||||
const first = await client.getCollabTokenWithReauth(); // caches token-1
|
||||
assert.equal(first, "provider-token-1");
|
||||
|
||||
// A forced refresh (what the reauth path passes) must NOT return the cached
|
||||
// token-1; it mints a fresh token-2 and replaces the cache.
|
||||
const forced = await client.getCollabTokenWithReauth(true);
|
||||
assert.equal(forced, "provider-token-2", "cache bypassed on forceRefresh");
|
||||
assert.equal(p.calls, 2);
|
||||
|
||||
const cached = await client.getCollabTokenWithReauth();
|
||||
assert.equal(cached, "provider-token-2", "the fresh token replaced the cache");
|
||||
assert.equal(p.calls, 2);
|
||||
});
|
||||
|
||||
test("two consecutive mutations keep the SAME token, so the session key is stable (provider path)", async () => {
|
||||
// The whole point of #435: acquireCollabSession keys on the token, so two
|
||||
// acquire calls in a burst must be handed the identical token string.
|
||||
process.env[ENV_KEY] = "300000";
|
||||
const p = countingProvider();
|
||||
const client = new DocmostClient({
|
||||
apiUrl: "http://127.0.0.1:1/api",
|
||||
getToken: async () => "access",
|
||||
getCollabToken: p.fn,
|
||||
});
|
||||
|
||||
const t1 = await client.getCollabTokenWithReauth();
|
||||
const t2 = await client.getCollabTokenWithReauth();
|
||||
assert.equal(t1, t2, "identical token across two mutations -> one session key");
|
||||
assert.equal(p.calls, 1);
|
||||
});
|
||||
|
||||
// ===========================================================================
|
||||
// REST /auth/collab-token path (external MCP)
|
||||
// ===========================================================================
|
||||
|
||||
test("within TTL, the REST /auth/collab-token endpoint is hit ONCE", async () => {
|
||||
process.env[ENV_KEY] = "300000";
|
||||
const state = { collabCalls: 0, loginCalls: 0 };
|
||||
const baseURL = await spawnCollabServer(state);
|
||||
const client = new DocmostClient(baseURL, "user@example.com", "pw");
|
||||
|
||||
const a = await client.getCollabTokenWithReauth();
|
||||
const b = await client.getCollabTokenWithReauth();
|
||||
|
||||
assert.equal(a, "collab-1");
|
||||
assert.equal(b, a, "cached token reused");
|
||||
assert.equal(state.collabCalls, 1, "POST /auth/collab-token called once");
|
||||
});
|
||||
|
||||
test("TTL=0 hits the REST endpoint on every call", async () => {
|
||||
process.env[ENV_KEY] = "0";
|
||||
const state = { collabCalls: 0, loginCalls: 0 };
|
||||
const baseURL = await spawnCollabServer(state);
|
||||
const client = new DocmostClient(baseURL, "user@example.com", "pw");
|
||||
|
||||
await client.getCollabTokenWithReauth();
|
||||
await client.getCollabTokenWithReauth();
|
||||
|
||||
assert.equal(state.collabCalls, 2, "cache disabled -> fetch each call");
|
||||
});
|
||||
|
||||
test("401 on REST collab-token re-logs-in and refetches (cache bypassed)", async () => {
|
||||
process.env[ENV_KEY] = "300000";
|
||||
const state = { collabCalls: 0, loginCalls: 0 };
|
||||
// The first collab-token mint 401s; the reauth path logs in and retries.
|
||||
const baseURL = await spawnCollabServer(state, { collabAuthFailsFor: 1 });
|
||||
const client = new DocmostClient(baseURL, "user@example.com", "pw");
|
||||
// Pre-seed a token so the initial call does not perform an initial login.
|
||||
client.token = "seed";
|
||||
client.client.defaults.headers.common["Authorization"] = "Bearer seed";
|
||||
|
||||
const tok = await client.getCollabTokenWithReauth();
|
||||
assert.equal(tok, "collab-2", "the post-reauth mint wins, not the failed one");
|
||||
assert.equal(state.loginCalls, 1, "re-login happened exactly once");
|
||||
assert.equal(state.collabCalls, 2, "one failed mint + one successful retry");
|
||||
});
|
||||
|
||||
test("a fresh login clears the cache so a collab token cannot outlive the identity", async () => {
|
||||
process.env[ENV_KEY] = "300000";
|
||||
const state = { collabCalls: 0, loginCalls: 0 };
|
||||
const baseURL = await spawnCollabServer(state);
|
||||
const client = new DocmostClient(baseURL, "user@example.com", "pw");
|
||||
|
||||
const before = await client.getCollabTokenWithReauth();
|
||||
assert.equal(before, "collab-1");
|
||||
|
||||
// Simulate an identity change (the 401 interceptor / re-login path calls
|
||||
// login(), which must drop the cached collab token).
|
||||
await client.login();
|
||||
|
||||
const after = await client.getCollabTokenWithReauth();
|
||||
assert.equal(after, "collab-2", "cache was invalidated by login(); refetched");
|
||||
assert.equal(state.collabCalls, 2);
|
||||
});
|
||||
Reference in New Issue
Block a user