6bf8361936
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) <noreply@anthropic.com>
770 lines
28 KiB
TypeScript
770 lines
28 KiB
TypeScript
import {
|
|
BadRequestException,
|
|
ConflictException,
|
|
ForbiddenException,
|
|
Inject,
|
|
Injectable,
|
|
InternalServerErrorException,
|
|
Logger,
|
|
NotFoundException,
|
|
} from '@nestjs/common';
|
|
import { InjectQueue } from '@nestjs/bullmq';
|
|
import { Queue } from 'bullmq';
|
|
import { CreateCommentDto, yjsSelectionSchema } from './dto/create-comment.dto';
|
|
import { CollaborationGateway } from '../../collaboration/collaboration.gateway';
|
|
import { UpdateCommentDto } from './dto/update-comment.dto';
|
|
import { CommentRepo } from '@docmost/db/repos/comment/comment.repo';
|
|
import { Comment, Page, User } from '@docmost/db/types/entity.types';
|
|
import { PaginationOptions } from '@docmost/db/pagination/pagination-options';
|
|
import { PageRepo } from '@docmost/db/repos/page/page.repo';
|
|
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';
|
|
import { WsService } from '../../ws/ws.service';
|
|
import {
|
|
AuthProvenanceData,
|
|
agentSourceFields,
|
|
} from '../../common/decorators/auth-provenance.decorator';
|
|
import { AuditEvent, AuditResource } from '../../common/events/audit-events';
|
|
import {
|
|
AUDIT_SERVICE,
|
|
IAuditService,
|
|
} from '../../integrations/audit/audit.service';
|
|
|
|
// Ephemeral-suggestion settle result (#329): 'deleted' → the comment vanished
|
|
// (hard-delete + anchor mark stripped); 'resolved' → the thread had replies and
|
|
// was resolved instead. Returned to the client so it can pick the optimistic
|
|
// cache action.
|
|
export type SuggestionOutcome = 'deleted' | 'resolved';
|
|
|
|
@Injectable()
|
|
export class CommentService {
|
|
private readonly logger = new Logger(CommentService.name);
|
|
|
|
constructor(
|
|
private commentRepo: CommentRepo,
|
|
private pageRepo: PageRepo,
|
|
private wsService: WsService,
|
|
private collaborationGateway: CollaborationGateway,
|
|
@InjectQueue(QueueName.GENERAL_QUEUE)
|
|
private generalQueue: Queue,
|
|
@InjectQueue(QueueName.NOTIFICATION_QUEUE)
|
|
private notificationQueue: Queue,
|
|
@Inject(AUDIT_SERVICE) private auditService: IAuditService,
|
|
) {}
|
|
|
|
async findById(commentId: string) {
|
|
const comment = await this.commentRepo.findById(commentId, {
|
|
includeCreator: true,
|
|
includeResolvedBy: true,
|
|
});
|
|
if (!comment) {
|
|
throw new NotFoundException('Comment not found');
|
|
}
|
|
return comment;
|
|
}
|
|
|
|
async create(
|
|
opts: { page: Page; workspaceId: string; user: User },
|
|
createCommentDto: CreateCommentDto,
|
|
// Optional agent-edit provenance (from the signed access claim). When the
|
|
// actor is 'agent', stamp created_source/ai_chat_id so an agent-authored
|
|
// comment (incl. a reply) shows the AI marker (§15 C3). Normal user: default.
|
|
provenance?: AuthProvenanceData,
|
|
) {
|
|
const { page, workspaceId, user } = opts;
|
|
const commentContent = JSON.parse(createCommentDto.content);
|
|
|
|
if (createCommentDto.parentCommentId) {
|
|
const parentComment = await this.commentRepo.findById(
|
|
createCommentDto.parentCommentId,
|
|
);
|
|
|
|
if (!parentComment || parentComment.pageId !== page.id) {
|
|
throw new BadRequestException('Parent comment not found');
|
|
}
|
|
|
|
if (parentComment.parentCommentId !== null) {
|
|
throw new BadRequestException('You cannot reply to a reply');
|
|
}
|
|
}
|
|
|
|
// Do NOT lossily truncate at 250: for a suggestion the client sends the RAW
|
|
// anchored document substring (the exact text under the comment mark) as the
|
|
// selection, which can be LONGER than the agent's <=250-char typed input
|
|
// (normalization collapses whitespace/typographic runs, so the raw span can
|
|
// exceed the normalized selection). Truncating it shorter than the mark span
|
|
// would break the apply-time equality check and make the suggestion
|
|
// un-appliable. Keep a generous 2000-char safety bound (matching
|
|
// suggestedText) so a legitimate anchored substring is never cut.
|
|
const selection = createCommentDto?.selection?.substring(0, 2000) ?? null;
|
|
|
|
// A suggested edit rewrites the exact text under an inline comment mark, so
|
|
// it is only meaningful on a top-level inline comment that carries a
|
|
// selection, and only if the suggestion actually changes that text.
|
|
let suggestedText: string | null = null;
|
|
if (
|
|
createCommentDto.suggestedText !== undefined &&
|
|
createCommentDto.suggestedText !== null
|
|
) {
|
|
if (createCommentDto.parentCommentId) {
|
|
throw new BadRequestException(
|
|
'A suggested edit can only be attached to a top-level comment, not a reply',
|
|
);
|
|
}
|
|
if (!selection || selection.trim().length === 0) {
|
|
throw new BadRequestException(
|
|
'A suggested edit requires an inline comment with a non-empty text selection',
|
|
);
|
|
}
|
|
const trimmed = createCommentDto.suggestedText.trim();
|
|
if (trimmed.length === 0) {
|
|
throw new BadRequestException('A suggested edit cannot be empty');
|
|
}
|
|
// A no-op suggestion (identical to the selection) is meaningless and would
|
|
// make "apply" indistinguishable from "already applied".
|
|
if (trimmed === selection.trim()) {
|
|
throw new BadRequestException(
|
|
'A suggested edit must differ from the selected text',
|
|
);
|
|
}
|
|
suggestedText = trimmed;
|
|
}
|
|
|
|
const inserted = await this.commentRepo.insertComment({
|
|
pageId: page.id,
|
|
content: commentContent,
|
|
selection,
|
|
type: createCommentDto.type ?? 'page',
|
|
parentCommentId: createCommentDto?.parentCommentId,
|
|
creatorId: user.id,
|
|
workspaceId: workspaceId,
|
|
spaceId: page.spaceId,
|
|
suggestedText,
|
|
// Agent-edit provenance: the user stays creatorId; this only annotates the
|
|
// source. Normal user requests leave the column default ('user').
|
|
...agentSourceFields(provenance, 'createdSource', 'aiChatId'),
|
|
});
|
|
|
|
if (createCommentDto.yjsSelection) {
|
|
const parsed = yjsSelectionSchema.safeParse(createCommentDto.yjsSelection);
|
|
if (!parsed.success) {
|
|
this.logger.warn(
|
|
`Invalid yjsSelection for comment ${inserted.id}: ${parsed.error.message}`,
|
|
);
|
|
} else {
|
|
const documentName = `page.${page.id}`;
|
|
try {
|
|
await this.collaborationGateway.handleYjsEvent(
|
|
'setCommentMark',
|
|
documentName,
|
|
{
|
|
yjsSelection: parsed.data,
|
|
commentId: inserted.id,
|
|
resolved: false,
|
|
user,
|
|
},
|
|
);
|
|
} catch (error) {
|
|
this.logger.warn(
|
|
`Failed to apply comment mark for comment ${inserted.id}, comment saved without inline highlight`,
|
|
error,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
const comment = await this.commentRepo.findById(inserted.id, {
|
|
includeCreator: true,
|
|
includeResolvedBy: true,
|
|
});
|
|
|
|
this.generalQueue
|
|
.add(QueueJob.ADD_PAGE_WATCHERS, {
|
|
userIds: [user.id],
|
|
pageId: page.id,
|
|
spaceId: page.spaceId,
|
|
workspaceId,
|
|
})
|
|
.catch((err) =>
|
|
this.logger.warn(`Failed to queue add-page-watchers: ${err.message}`),
|
|
);
|
|
|
|
const isReply = !!createCommentDto.parentCommentId;
|
|
|
|
await this.queueCommentNotification(
|
|
commentContent,
|
|
[],
|
|
comment.id,
|
|
page.id,
|
|
page.spaceId,
|
|
workspaceId,
|
|
user.id,
|
|
!isReply,
|
|
createCommentDto.parentCommentId,
|
|
);
|
|
|
|
this.wsService.emitCommentEvent(page.spaceId, page.id, {
|
|
operation: 'commentCreated',
|
|
pageId: page.id,
|
|
comment,
|
|
});
|
|
|
|
return comment;
|
|
}
|
|
|
|
async findByPageId(
|
|
pageId: string,
|
|
pagination: PaginationOptions,
|
|
): Promise<CursorPaginationResult<Comment>> {
|
|
const page = await this.pageRepo.findById(pageId);
|
|
|
|
if (!page) {
|
|
throw new BadRequestException('Page not found');
|
|
}
|
|
|
|
return this.commentRepo.findPageComments(pageId, pagination);
|
|
}
|
|
|
|
async update(
|
|
comment: Comment,
|
|
updateCommentDto: UpdateCommentDto,
|
|
authUser: User,
|
|
): Promise<Comment> {
|
|
const commentContent = JSON.parse(updateCommentDto.content);
|
|
|
|
if (comment.creatorId !== authUser.id) {
|
|
throw new ForbiddenException('You can only edit your own comments');
|
|
}
|
|
|
|
const oldMentionIds = extractUserMentionIdsFromJson(comment.content);
|
|
|
|
const editedAt = new Date();
|
|
|
|
await this.commentRepo.updateComment(
|
|
{
|
|
content: commentContent,
|
|
editedAt: editedAt,
|
|
updatedAt: editedAt,
|
|
},
|
|
comment.id,
|
|
);
|
|
|
|
await this.queueCommentNotification(
|
|
commentContent,
|
|
oldMentionIds,
|
|
comment.id,
|
|
comment.pageId,
|
|
comment.spaceId,
|
|
comment.workspaceId,
|
|
authUser.id,
|
|
false,
|
|
);
|
|
|
|
// Re-fetch the enriched comment before broadcasting, symmetric with
|
|
// create()/resolveComment(). updateComment() above has already persisted the
|
|
// new content/timestamps, so this single-row read reflects the edit AND
|
|
// carries the same {agent,launcher} avatar stack (via includeCreator) as the
|
|
// other two broadcasts. This deliberately does NOT reuse the caller's
|
|
// pre-loaded `comment`: relying on the controller happening to load it with
|
|
// includeCreator:true is exactly the fragile coupling that let the agent
|
|
// stack silently vanish on edit once already (#300/#304) — a future caller
|
|
// dropping that flag must not regress the broadcast.
|
|
const updatedComment = await this.commentRepo.findById(comment.id, {
|
|
includeCreator: true,
|
|
includeResolvedBy: true,
|
|
});
|
|
|
|
this.wsService.emitCommentEvent(comment.spaceId, comment.pageId, {
|
|
operation: 'commentUpdated',
|
|
pageId: comment.pageId,
|
|
comment: updatedComment,
|
|
});
|
|
|
|
return updatedComment;
|
|
}
|
|
|
|
async resolveComment(
|
|
comment: Comment,
|
|
resolved: boolean,
|
|
authUser: User,
|
|
// Optional agent-edit provenance (from the signed access claim). When the
|
|
// actor is 'agent' and the thread is being resolved, stamp resolved_source
|
|
// so the "resolved by" mark shows the AI marker (§15 C3). On unresolve the
|
|
// source is cleared alongside resolvedAt/resolvedById.
|
|
provenance?: AuthProvenanceData,
|
|
): Promise<Comment> {
|
|
// 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
|
|
// null so a reopened thread carries no stale source. A normal user resolve
|
|
// leaves resolved_source null (no agent annotation).
|
|
const resolvedSource = resolved && isAgent ? 'agent' : null;
|
|
|
|
await this.commentRepo.updateComment(
|
|
// 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,
|
|
);
|
|
|
|
// #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}`;
|
|
void this.enqueueCommentMarkUpdate(
|
|
documentName,
|
|
comment.id,
|
|
resolved ? 'resolve' : 'unresolve',
|
|
now.getTime(),
|
|
authUser.id,
|
|
).catch((error) =>
|
|
this.logger.warn(
|
|
`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) {
|
|
const jobData: ICommentResolvedNotificationJob = {
|
|
commentId: comment.id,
|
|
commentCreatorId: comment.creatorId,
|
|
pageId: comment.pageId,
|
|
spaceId: comment.spaceId,
|
|
workspaceId: comment.workspaceId,
|
|
actorId: authUser.id,
|
|
};
|
|
await this.notificationQueue.add(
|
|
QueueJob.COMMENT_RESOLVED_NOTIFICATION,
|
|
jobData,
|
|
);
|
|
}
|
|
|
|
const updatedComment = await this.commentRepo.findById(comment.id, {
|
|
includeCreator: true,
|
|
includeResolvedBy: true,
|
|
});
|
|
|
|
this.wsService.emitCommentEvent(comment.spaceId, comment.pageId, {
|
|
operation: 'commentResolved',
|
|
pageId: comment.pageId,
|
|
comment: updatedComment,
|
|
});
|
|
|
|
return updatedComment;
|
|
}
|
|
|
|
/**
|
|
* Apply the suggested edit carried by a top-level inline comment: atomically
|
|
* replace the text under the comment mark in the collaborative document with
|
|
* the comment's suggestedText, then stamp the applied fields and auto-resolve
|
|
* the thread. The controller authorizes (validateCanEdit); this re-checks the
|
|
* comment's own state so the invariant holds regardless of caller.
|
|
*/
|
|
async applySuggestion(
|
|
comment: Comment,
|
|
user: User,
|
|
provenance?: AuthProvenanceData,
|
|
): Promise<Comment & { outcome: SuggestionOutcome }> {
|
|
// Structural guards.
|
|
if (comment.parentCommentId) {
|
|
throw new BadRequestException(
|
|
'Only a top-level comment can carry a suggested edit',
|
|
);
|
|
}
|
|
if (!comment.suggestedText) {
|
|
throw new BadRequestException('This comment has no suggested edit to apply');
|
|
}
|
|
// State guards. Order matters — the already-applied check precedes the
|
|
// resolved check because an applied comment is normally also resolved.
|
|
//
|
|
// Already applied → IDEMPOTENT SUCCESS (issue #315 DoD: double-click /
|
|
// two-user race → idempotent "already applied", NOT a 409). The suggestion
|
|
// is already in the document, so do NOT call the collab gateway again.
|
|
// finalizeAppliedSuggestion re-fetches/broadcasts the same success shape as
|
|
// the applied branch and, when the thread is still open (the rare "applied
|
|
// but not resolved" crash window), self-heals it via resolveComment.
|
|
if (comment.suggestionAppliedAt) {
|
|
return this.finalizeAppliedSuggestion(comment, user, provenance);
|
|
}
|
|
// Not-yet-applied on a resolved thread → reject. The client hides the apply
|
|
// button once a thread is resolved; this is the defensive server check.
|
|
if (comment.resolvedAt) {
|
|
throw new BadRequestException(
|
|
'Cannot apply a suggested edit on a resolved comment thread',
|
|
);
|
|
}
|
|
|
|
// Derive the document name the same way create()/resolveComment() do for
|
|
// the comment marks: `page.${pageId}`.
|
|
const documentName = `page.${comment.pageId}`;
|
|
|
|
let verdict: { applied: boolean; currentText: string | null } | undefined;
|
|
try {
|
|
verdict = await this.collaborationGateway.handleYjsEvent(
|
|
'applyCommentSuggestion',
|
|
documentName,
|
|
{
|
|
commentId: comment.id,
|
|
expectedText: comment.selection,
|
|
newText: comment.suggestedText,
|
|
user,
|
|
},
|
|
);
|
|
} catch (error) {
|
|
// A throwing gateway (or the phase-3 fallback failing) is a hard error —
|
|
// never silently succeed, the document may or may not have changed.
|
|
this.logger.error(
|
|
`Failed to apply suggested edit for comment ${comment.id}`,
|
|
error,
|
|
);
|
|
throw new InternalServerErrorException('Failed to apply the suggested edit');
|
|
}
|
|
|
|
if (!verdict) {
|
|
// Should not happen given the phase-3 fallback; treat as a hard error
|
|
// rather than assuming success.
|
|
throw new InternalServerErrorException('Failed to apply the suggested edit');
|
|
}
|
|
|
|
if (verdict.applied === true) {
|
|
return this.finalizeAppliedSuggestion(comment, user, provenance);
|
|
}
|
|
|
|
// Idempotent branch: the mutation didn't run now, but the text under the
|
|
// mark is ALREADY the suggested text (double-click, two-user race, or a
|
|
// crash between the doc mutation and the DB write). Reconcile the DB /
|
|
// resolved state and report success — do NOT 409.
|
|
if (
|
|
verdict.applied === false &&
|
|
verdict.currentText === comment.suggestedText
|
|
) {
|
|
return this.finalizeAppliedSuggestion(comment, user, provenance);
|
|
}
|
|
|
|
// The commented text changed since the suggestion was made. Surface the
|
|
// current text so the client can tell the user what it is now.
|
|
throw new ConflictException({
|
|
message:
|
|
'The commented text changed since this suggestion was made; it was not applied.',
|
|
currentText: verdict.currentText,
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Dismiss ("Не применять") a suggested edit without touching the page text:
|
|
* the suggestion disappears. Ephemeral rule (#329) — a top-level suggestion
|
|
* comment is transient UI, so dismissing it hard-deletes the comment AND strips
|
|
* its inline anchor mark UNLESS the thread has replies, in which case the
|
|
* discussion is preserved by resolving it instead.
|
|
*
|
|
* Dismiss does NOT change the document text, so the controller authorizes it
|
|
* with canComment (NOT canEdit). This re-checks the comment's own state so the
|
|
* invariant holds regardless of caller.
|
|
*/
|
|
async dismissSuggestion(
|
|
comment: Comment,
|
|
user: User,
|
|
provenance?: AuthProvenanceData,
|
|
): Promise<Comment & { outcome: SuggestionOutcome }> {
|
|
// Structural guards (mirror applySuggestion).
|
|
if (comment.parentCommentId) {
|
|
throw new BadRequestException(
|
|
'Only a top-level comment can carry a suggested edit',
|
|
);
|
|
}
|
|
if (!comment.suggestedText) {
|
|
throw new BadRequestException(
|
|
'This comment has no suggested edit to dismiss',
|
|
);
|
|
}
|
|
// State guards: dismissing an already-applied or already-resolved thread is
|
|
// meaningless. On an apply↔dismiss race the loser sees the comment already
|
|
// gone (404 at the controller) or already resolved (this 400); the client
|
|
// treats both as "already resolved".
|
|
if (comment.suggestionAppliedAt) {
|
|
throw new BadRequestException(
|
|
'Cannot dismiss a suggested edit that was already applied',
|
|
);
|
|
}
|
|
if (comment.resolvedAt) {
|
|
throw new BadRequestException(
|
|
'Cannot dismiss a suggested edit on a resolved comment thread',
|
|
);
|
|
}
|
|
|
|
const hasChildren = await this.commentRepo.hasChildren(comment.id);
|
|
|
|
if (hasChildren) {
|
|
// Preserve the discussion: resolve (never delete) a thread with replies.
|
|
const updatedComment = await this.resolveComment(
|
|
comment,
|
|
true,
|
|
user,
|
|
provenance,
|
|
);
|
|
this.auditService.log({
|
|
event: AuditEvent.COMMENT_SUGGESTION_DISMISSED,
|
|
resourceType: AuditResource.COMMENT,
|
|
resourceId: comment.id,
|
|
spaceId: comment.spaceId,
|
|
metadata: { pageId: comment.pageId },
|
|
});
|
|
return { ...updatedComment, outcome: 'resolved' };
|
|
}
|
|
|
|
// Ephemeral: no replies → the suggestion vanishes entirely. The atomic
|
|
// conditional delete may still fall back to a resolve if a reply raced in
|
|
// (see deleteEphemeralSuggestion), so the outcome is whatever it settled on.
|
|
const settled = await this.deleteEphemeralSuggestion(comment, user, provenance);
|
|
this.auditService.log({
|
|
event: AuditEvent.COMMENT_SUGGESTION_DISMISSED,
|
|
resourceType: AuditResource.COMMENT,
|
|
resourceId: comment.id,
|
|
spaceId: comment.spaceId,
|
|
metadata: { pageId: comment.pageId },
|
|
});
|
|
return settled;
|
|
}
|
|
|
|
/**
|
|
* Persist the applied stamps (idempotently), then settle the suggestion under
|
|
* the ephemeral rule (#329): a suggestion whose thread has NO replies
|
|
* DISAPPEARS after apply (hard-delete + strip the inline anchor mark), since
|
|
* the suggested text is now in the document and a stand-alone resolved thread
|
|
* would only pile up an orphan anchor. A thread WITH replies is preserved by
|
|
* auto-resolving it (the historical behaviour). Shared by the applied and the
|
|
* idempotent "already-applied" branches of applySuggestion.
|
|
*
|
|
* Returns the comment augmented with `outcome` so the client can pick the
|
|
* optimistic action ('deleted' → drop it, 'resolved' → move to the resolved
|
|
* tab).
|
|
*/
|
|
private async finalizeAppliedSuggestion(
|
|
comment: Comment,
|
|
user: User,
|
|
provenance?: AuthProvenanceData,
|
|
): Promise<Comment & { outcome: SuggestionOutcome }> {
|
|
const hasChildren = await this.commentRepo.hasChildren(comment.id);
|
|
|
|
if (hasChildren) {
|
|
// Thread has replies → preserve the discussion: stamp applied + resolve.
|
|
if (!comment.suggestionAppliedAt) {
|
|
await this.commentRepo.updateComment(
|
|
{
|
|
suggestionAppliedAt: new Date(),
|
|
suggestionAppliedById: user.id,
|
|
},
|
|
comment.id,
|
|
);
|
|
}
|
|
|
|
// Auto-resolve the thread. resolveComment handles the resolve mark, its ws
|
|
// broadcast and the resolve notification. Stay defensive on re-entry.
|
|
if (!comment.resolvedAt) {
|
|
await this.resolveComment(comment, true, user, provenance);
|
|
}
|
|
|
|
const updatedComment = await this.commentRepo.findById(comment.id, {
|
|
includeCreator: true,
|
|
includeResolvedBy: true,
|
|
});
|
|
|
|
this.wsService.emitCommentEvent(comment.spaceId, comment.pageId, {
|
|
operation: 'commentUpdated',
|
|
pageId: comment.pageId,
|
|
comment: updatedComment,
|
|
});
|
|
|
|
this.auditService.log({
|
|
event: AuditEvent.COMMENT_SUGGESTION_APPLIED,
|
|
resourceType: AuditResource.COMMENT,
|
|
resourceId: comment.id,
|
|
spaceId: comment.spaceId,
|
|
metadata: { pageId: comment.pageId },
|
|
});
|
|
|
|
return { ...updatedComment, outcome: 'resolved' };
|
|
}
|
|
|
|
// No replies → ephemeral: the suggested text is already in the document, so
|
|
// the comment is redundant. Hard-delete it and strip its inline anchor. We
|
|
// deliberately do NOT write the applied stamps first (the row is about to be
|
|
// deleted); the audit event still records that the suggestion was applied.
|
|
// The delete is atomic-conditional: if a reply raced in after the
|
|
// hasChildren read, it falls back to resolving instead (outcome 'resolved').
|
|
const settled = await this.deleteEphemeralSuggestion(comment, user, provenance);
|
|
|
|
this.auditService.log({
|
|
event: AuditEvent.COMMENT_SUGGESTION_APPLIED,
|
|
resourceType: AuditResource.COMMENT,
|
|
resourceId: comment.id,
|
|
spaceId: comment.spaceId,
|
|
metadata: { pageId: comment.pageId },
|
|
});
|
|
|
|
return settled;
|
|
}
|
|
|
|
/**
|
|
* Settle an ephemeral suggestion whose thread looked childless: remove its
|
|
* inline `comment` anchor mark, then ATOMICALLY hard-delete the row only if it
|
|
* is still childless. Shared by the apply/dismiss no-replies branches (#329).
|
|
*
|
|
* ORDER MATTERS: the anchor mark is removed FIRST and FATALLY (mirrors
|
|
* applySuggestion, which mutates the doc before writing the DB). The row
|
|
* delete is irreversible, so if the mark removal fails — including the
|
|
* COLLAB_DISABLE_REDIS "no live instance" hard-error — we must NOT delete the
|
|
* row and report success, or the document is left with a permanent orphan
|
|
* anchor pointing at a comment that no longer exists (the exact data-integrity
|
|
* bug #329 targets). Let the exception propagate (→ 5xx); the operation is
|
|
* then repeatable with row + mark still consistent.
|
|
*
|
|
* RACE (#338 F4): the caller read `hasChildren` BEFORE the (slow) mark
|
|
* removal, so a reply can land in that window. `comments.parent_comment_id` is
|
|
* ON DELETE CASCADE, so an unconditional delete here would cascade-destroy the
|
|
* just-added reply forever. Instead we use `deleteCommentIfChildless`, which
|
|
* re-checks childlessness under a FOR UPDATE lock inside a transaction (a plain
|
|
* anti-join DELETE is NOT race-safe under READ COMMITTED — see the repo method
|
|
* docstring). If it removes the row (outcome 'deleted') we broadcast the
|
|
* deletion as before. If it removes 0 rows (a reply interleaved) we do NOT
|
|
* hard-delete — we resolve the thread instead (outcome 'resolved'), preserving
|
|
* the discussion and the new reply. The anchor mark is already gone by then, an
|
|
* accepted degradation: the thread lands in the resolved tab without its inline
|
|
* highlight — far better than losing a reply.
|
|
*/
|
|
private async deleteEphemeralSuggestion(
|
|
comment: Comment,
|
|
user: User,
|
|
provenance?: AuthProvenanceData,
|
|
): Promise<Comment & { outcome: SuggestionOutcome }> {
|
|
await this.deleteCommentMark(comment, user);
|
|
|
|
const deletedRows = await this.commentRepo.deleteCommentIfChildless(
|
|
comment.id,
|
|
);
|
|
|
|
if (deletedRows > 0) {
|
|
this.wsService.emitCommentEvent(comment.spaceId, comment.pageId, {
|
|
operation: 'commentDeleted',
|
|
pageId: comment.pageId,
|
|
commentId: comment.id,
|
|
});
|
|
return { ...comment, outcome: 'deleted' };
|
|
}
|
|
|
|
// A reply interleaved between the hasChildren read and this delete, so the
|
|
// conditional delete matched nothing. Preserve the discussion + the new
|
|
// reply by resolving the thread instead of hard-deleting it. resolveComment
|
|
// handles the resolve patch, its ws broadcast and the resolve notification;
|
|
// its collab call is best-effort, so the already-stripped mark is fine.
|
|
const resolvedComment = await this.resolveComment(
|
|
comment,
|
|
true,
|
|
user,
|
|
provenance,
|
|
);
|
|
return { ...resolvedComment, outcome: 'resolved' };
|
|
}
|
|
|
|
/**
|
|
* 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.enqueueCommentMarkUpdate(
|
|
documentName,
|
|
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[],
|
|
commentId: string,
|
|
pageId: string,
|
|
spaceId: string,
|
|
workspaceId: string,
|
|
actorId: string,
|
|
notifyWatchers: boolean,
|
|
parentCommentId?: string,
|
|
) {
|
|
const mentionedUserIds = extractUserMentionIdsFromJson(content);
|
|
const newMentionIds = mentionedUserIds.filter(
|
|
(id) => id !== actorId && !oldMentionIds.includes(id),
|
|
);
|
|
|
|
if (newMentionIds.length === 0 && !notifyWatchers && !parentCommentId) return;
|
|
|
|
const jobData: ICommentNotificationJob = {
|
|
commentId,
|
|
parentCommentId,
|
|
pageId,
|
|
spaceId,
|
|
workspaceId,
|
|
actorId,
|
|
mentionedUserIds: newMentionIds,
|
|
notifyWatchers,
|
|
};
|
|
|
|
await this.notificationQueue.add(
|
|
QueueJob.COMMENT_NOTIFICATION,
|
|
jobData,
|
|
);
|
|
}
|
|
}
|