diff --git a/apps/client/src/features/page-history/components/history-item.tsx b/apps/client/src/features/page-history/components/history-item.tsx index e44614c4..cc56b191 100644 --- a/apps/client/src/features/page-history/components/history-item.tsx +++ b/apps/client/src/features/page-history/components/history-item.tsx @@ -1,4 +1,4 @@ -import { Text, Group, UnstyledButton } from "@mantine/core"; +import { Text, Group, UnstyledButton, Avatar, Tooltip } from "@mantine/core"; import { CustomAvatar } from "@/components/ui/custom-avatar.tsx"; import { formattedDate } from "@/lib/time"; import classes from "./css/history.module.css"; @@ -6,6 +6,8 @@ import clsx from "clsx"; import { IPageHistory } from "@/features/page-history/types/page.types"; import { memo, useCallback } from "react"; +const MAX_VISIBLE_AVATARS = 5; + interface HistoryItemProps { historyItem: IPageHistory; index: number; @@ -31,6 +33,9 @@ const HistoryItem = memo(function HistoryItem({ onHover?.(historyItem.id, index); }, [onHover, historyItem.id, index]); + const contributors = historyItem.contributors; + const hasContributors = contributors && contributors.length > 0; + return ( - -
- - {formattedDate(new Date(historyItem.createdAt))} - + {formattedDate(new Date(historyItem.createdAt))} -
- - + + {hasContributors ? ( + <> + + + {contributors.slice(0, MAX_VISIBLE_AVATARS).map((contributor) => ( + + + + ))} + {contributors.length > MAX_VISIBLE_AVATARS && ( + ( +
{c.name}
+ ))} + > + + +{contributors.length - MAX_VISIBLE_AVATARS} + + + )} + + + {contributors.length === 1 && ( - {historyItem.lastUpdatedBy?.name} + {contributors[0].name} - -
-
+ )} + + ) : ( + <> + + + {historyItem.lastUpdatedBy?.name} + + + )} ); diff --git a/apps/client/src/features/page-history/components/history-modal-mobile.tsx b/apps/client/src/features/page-history/components/history-modal-mobile.tsx index 0a54a01c..b73695da 100644 --- a/apps/client/src/features/page-history/components/history-modal-mobile.tsx +++ b/apps/client/src/features/page-history/components/history-modal-mobile.tsx @@ -62,11 +62,18 @@ export default function HistoryModalMobile({ pageId, pageTitle }: Props) { const selectData = useMemo( () => - historyItems.map((item) => ({ - value: item.id, - label: formattedDate(new Date(item.createdAt)), - userName: item.lastUpdatedBy?.name, - })), + historyItems.map((item) => { + const contributors = item.contributors; + const hasContributors = contributors && contributors.length > 0; + const names = hasContributors + ? contributors.map((c) => c.name).join(", ") + : item.lastUpdatedBy?.name; + return { + value: item.id, + label: formattedDate(new Date(item.createdAt)), + userName: names, + }; + }), [historyItems], ); diff --git a/apps/client/src/features/page-history/types/page.types.ts b/apps/client/src/features/page-history/types/page.types.ts index 98a5b9d7..45b2e1f8 100644 --- a/apps/client/src/features/page-history/types/page.types.ts +++ b/apps/client/src/features/page-history/types/page.types.ts @@ -18,4 +18,5 @@ export interface IPageHistory { createdAt: string; updatedAt: string; lastUpdatedBy: IPageHistoryUser; + contributors?: IPageHistoryUser[]; } diff --git a/apps/server/src/collaboration/collaboration.module.ts b/apps/server/src/collaboration/collaboration.module.ts index e9374c53..c4444d0b 100644 --- a/apps/server/src/collaboration/collaboration.module.ts +++ b/apps/server/src/collaboration/collaboration.module.ts @@ -7,9 +7,10 @@ import { CollabWsAdapter } from './adapter/collab-ws.adapter'; import { IncomingMessage } from 'http'; import { WebSocket } from 'ws'; import { TokenModule } from '../core/auth/token.module'; -import { HistoryListener } from './listeners/history.listener'; +import { HistoryProcessor } from './processors/history.processor'; import { LoggerExtension } from './extensions/logger.extension'; import { CollaborationHandler } from './collaboration.handler'; +import { CollabHistoryService } from './services/collab-history.service'; @Module({ providers: [ @@ -17,7 +18,8 @@ import { CollaborationHandler } from './collaboration.handler'; AuthenticationExtension, PersistenceExtension, LoggerExtension, - HistoryListener, + HistoryProcessor, + CollabHistoryService, CollaborationHandler, ], exports: [CollaborationGateway], diff --git a/apps/server/src/collaboration/constants.ts b/apps/server/src/collaboration/constants.ts new file mode 100644 index 00000000..8ce8c825 --- /dev/null +++ b/apps/server/src/collaboration/constants.ts @@ -0,0 +1,3 @@ +export const HISTORY_INTERVAL = 5 * 60 * 1000; +export const HISTORY_FAST_INTERVAL = 60 * 1000; +export const HISTORY_FAST_THRESHOLD = 5 * 60 * 1000; diff --git a/apps/server/src/collaboration/extensions/persistence.extension.ts b/apps/server/src/collaboration/extensions/persistence.extension.ts index 54c4a89e..4548b40c 100644 --- a/apps/server/src/collaboration/extensions/persistence.extension.ts +++ b/apps/server/src/collaboration/extensions/persistence.extension.ts @@ -13,7 +13,6 @@ import { PageRepo } from '@docmost/db/repos/page/page.repo'; import { InjectKysely } from 'nestjs-kysely'; import { KyselyDB } from '@docmost/db/types/kysely.types'; import { executeTx } from '@docmost/db/utils'; -import { EventEmitter2 } from '@nestjs/event-emitter'; import { InjectQueue } from '@nestjs/bullmq'; import { QueueJob, QueueName } from '../../integrations/queue/constants'; import { Queue } from 'bullmq'; @@ -22,8 +21,17 @@ import { extractPageMentions, } from '../../common/helpers/prosemirror/utils'; import { isDeepStrictEqual } from 'node:util'; -import { IPageBacklinkJob } from '../../integrations/queue/constants/queue.interface'; +import { + IPageBacklinkJob, + IPageHistoryJob, +} from '../../integrations/queue/constants/queue.interface'; import { Page } from '@docmost/db/types/entity.types'; +import { CollabHistoryService } from '../services/collab-history.service'; +import { + HISTORY_FAST_INTERVAL, + HISTORY_FAST_THRESHOLD, + HISTORY_INTERVAL, +} from '../constants'; @Injectable() export class PersistenceExtension implements Extension { @@ -33,9 +41,10 @@ export class PersistenceExtension implements Extension { constructor( private readonly pageRepo: PageRepo, @InjectKysely() private readonly db: KyselyDB, - private eventEmitter: EventEmitter2, @InjectQueue(QueueName.GENERAL_QUEUE) private generalQueue: Queue, @InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue, + @InjectQueue(QueueName.HISTORY_QUEUE) private historyQueue: Queue, + private readonly collabHistory: CollabHistoryService, ) {} async onLoadDocument(data: onLoadDocumentPayload) { @@ -101,6 +110,7 @@ export class PersistenceExtension implements Extension { } let page: Page = null; + const editingUserIds = this.consumeContributors(documentName); try { await executeTx(this.db, async (trx) => { @@ -123,13 +133,9 @@ export class PersistenceExtension implements Extension { let contributorIds = undefined; try { const existingContributors = page.contributorIds || []; - const contributorSet = this.contributors.get(documentName); - contributorSet.add(page.creatorId); - const newContributors = [...contributorSet]; contributorIds = Array.from( - new Set([...existingContributors, ...newContributors]), + new Set([...existingContributors, ...editingUserIds, page.creatorId]), ); - this.contributors.delete(documentName); } catch (err) { //this.logger.debug('Contributors error:' + err?.['message']); } @@ -153,13 +159,7 @@ export class PersistenceExtension implements Extension { } if (page) { - this.eventEmitter.emit('collab.page.updated', { - page: { - ...page, - content: tiptapJson, - lastUpdatedById: context.user.id, - }, - }); + await this.collabHistory.addContributors(pageId, editingUserIds); const mentions = extractMentions(tiptapJson); const pageMentions = extractPageMentions(mentions); @@ -174,6 +174,8 @@ export class PersistenceExtension implements Extension { pageIds: [pageId], workspaceId: page.workspaceId, }); + + await this.enqueuePageHistory(page); } } @@ -193,4 +195,26 @@ export class PersistenceExtension implements Extension { const documentName = data.documentName; this.contributors.delete(documentName); } + + private consumeContributors(documentName: string): string[] { + const contributorSet = this.contributors.get(documentName); + if (!contributorSet) return []; + const userIds = [...contributorSet]; + this.contributors.delete(documentName); + return userIds; + } + + private async enqueuePageHistory(page: Page): Promise { + const pageAge = Date.now() - new Date(page.createdAt).getTime(); + const delay = + pageAge < HISTORY_FAST_THRESHOLD + ? HISTORY_FAST_INTERVAL + : HISTORY_INTERVAL; + + await this.historyQueue.add( + QueueJob.PAGE_HISTORY, + { pageId: page.id } as IPageHistoryJob, + { jobId: page.id, delay }, + ); + } } diff --git a/apps/server/src/collaboration/listeners/history.listener.ts b/apps/server/src/collaboration/listeners/history.listener.ts deleted file mode 100644 index e0d40838..00000000 --- a/apps/server/src/collaboration/listeners/history.listener.ts +++ /dev/null @@ -1,52 +0,0 @@ -import { Injectable, Logger } from '@nestjs/common'; -import { OnEvent } from '@nestjs/event-emitter'; -import { PageHistoryRepo } from '@docmost/db/repos/page/page-history.repo'; -import { Page } from '@docmost/db/types/entity.types'; -import { isDeepStrictEqual } from 'node:util'; -import { EnvironmentService } from '../../integrations/environment/environment.service'; - -export class UpdatedPageEvent { - page: Page; -} - -@Injectable() -export class HistoryListener { - private readonly logger = new Logger(HistoryListener.name); - - constructor( - private readonly pageHistoryRepo: PageHistoryRepo, - private readonly environmentService: EnvironmentService, - ) {} - - @OnEvent('collab.page.updated') - async handleCreatePageHistory(event: UpdatedPageEvent) { - const { page } = event; - - const pageCreationTime = new Date(page.createdAt).getTime(); - const currentTime = Date.now(); - const FIVE_MINUTES = this.environmentService.isDevelopment() - ? 60 * 1000 - : 5 * 60 * 1000; - - if (currentTime - pageCreationTime < FIVE_MINUTES) { - return; - } - - const lastHistory = await this.pageHistoryRepo.findPageLastHistory(page.id, { - includeContent: true, - }); - - if ( - !lastHistory || - (!isDeepStrictEqual(lastHistory.content, page.content) && - currentTime - new Date(lastHistory.createdAt).getTime() >= FIVE_MINUTES) - ) { - try { - await this.pageHistoryRepo.saveHistory(page); - this.logger.debug(`New history created for: ${page.id}`); - } catch (err) { - this.logger.error(`Failed to create history for page: ${page.id}`, err); - } - } - } -} diff --git a/apps/server/src/collaboration/processors/history.processor.ts b/apps/server/src/collaboration/processors/history.processor.ts new file mode 100644 index 00000000..985f0b3e --- /dev/null +++ b/apps/server/src/collaboration/processors/history.processor.ts @@ -0,0 +1,84 @@ +import { Logger, OnModuleDestroy } from '@nestjs/common'; +import { OnWorkerEvent, Processor, WorkerHost } from '@nestjs/bullmq'; +import { Job } from 'bullmq'; +import { QueueJob, QueueName } from '../../integrations/queue/constants'; +import { IPageHistoryJob } from '../../integrations/queue/constants/queue.interface'; +import { PageHistoryRepo } from '@docmost/db/repos/page/page-history.repo'; +import { PageRepo } from '@docmost/db/repos/page/page.repo'; +import { isDeepStrictEqual } from 'node:util'; +import { CollabHistoryService } from '../services/collab-history.service'; + +@Processor(QueueName.HISTORY_QUEUE) +export class HistoryProcessor extends WorkerHost implements OnModuleDestroy { + private readonly logger = new Logger(HistoryProcessor.name); + + constructor( + private readonly pageHistoryRepo: PageHistoryRepo, + private readonly pageRepo: PageRepo, + private readonly collabHistory: CollabHistoryService, + ) { + super(); + } + + async process(job: Job): Promise { + if (job.name !== QueueJob.PAGE_HISTORY) return; + + try { + const { pageId } = job.data; + + const page = await this.pageRepo.findById(pageId, { + includeContent: true, + }); + + if (!page) { + this.logger.warn(`Page ${pageId} not found, skipping history`); + await this.collabHistory.clearContributors(pageId); + return; + } + + const lastHistory = await this.pageHistoryRepo.findPageLastHistory( + pageId, + { includeContent: true }, + ); + + if ( + !lastHistory || + !isDeepStrictEqual(lastHistory.content, page.content) + ) { + const contributorIds = + await this.collabHistory.popContributors(pageId); + + try { + await this.pageHistoryRepo.saveHistory(page, { contributorIds }); + this.logger.debug(`History created for page: ${pageId}`); + } catch (err) { + await this.collabHistory.addContributors( + pageId, + contributorIds, + ); + throw err; + } + } + } catch (err) { + throw err; + } + } + + @OnWorkerEvent('active') + onActive(job: Job) { + this.logger.debug(`Processing ${job.name} for page: ${job.data.pageId}`); + } + + @OnWorkerEvent('failed') + onError(job: Job) { + this.logger.error( + `Failed ${job.name} for page: ${job.data.pageId}. Reason: ${job.failedReason}`, + ); + } + + async onModuleDestroy(): Promise { + if (this.worker) { + await this.worker.close(); + } + } +} diff --git a/apps/server/src/collaboration/services/collab-history.service.ts b/apps/server/src/collaboration/services/collab-history.service.ts new file mode 100644 index 00000000..b7cf5086 --- /dev/null +++ b/apps/server/src/collaboration/services/collab-history.service.ts @@ -0,0 +1,30 @@ +import { Injectable } from '@nestjs/common'; +import { RedisService } from '@nestjs-labs/nestjs-ioredis'; +import type { Redis } from 'ioredis'; + +const REDIS_KEY_PREFIX = 'history:contributors:'; + +@Injectable() +export class CollabHistoryService { + private readonly redis: Redis; + + constructor(private readonly redisService: RedisService) { + this.redis = this.redisService.getOrThrow(); + } + + async addContributors(pageId: string, userIds: string[]): Promise { + if (userIds.length === 0) return; + await this.redis.sadd(REDIS_KEY_PREFIX + pageId, ...userIds); + } + + async popContributors(pageId: string): Promise { + const key = REDIS_KEY_PREFIX + pageId; + const count = await this.redis.scard(key); + if (count === 0) return []; + return await this.redis.spop(key, count); + } + + async clearContributors(pageId: string): Promise { + await this.redis.del(REDIS_KEY_PREFIX + pageId); + } +} diff --git a/apps/server/src/database/migrations/20260209T120000-add-contributor-ids-to-page-history.ts b/apps/server/src/database/migrations/20260209T120000-add-contributor-ids-to-page-history.ts new file mode 100644 index 00000000..4e50ad3a --- /dev/null +++ b/apps/server/src/database/migrations/20260209T120000-add-contributor-ids-to-page-history.ts @@ -0,0 +1,15 @@ +import { type Kysely, sql } from 'kysely'; + +export async function up(db: Kysely): Promise { + await db.schema + .alterTable('page_history') + .addColumn('contributor_ids', sql`uuid[]`, (col) => col.defaultTo('{}')) + .execute(); +} + +export async function down(db: Kysely): Promise { + await db.schema + .alterTable('page_history') + .dropColumn('contributor_ids') + .execute(); +} diff --git a/apps/server/src/database/repos/page/page-history.repo.ts b/apps/server/src/database/repos/page/page-history.repo.ts index c61c7e88..aca38f45 100644 --- a/apps/server/src/database/repos/page/page-history.repo.ts +++ b/apps/server/src/database/repos/page/page-history.repo.ts @@ -9,8 +9,8 @@ import { } from '@docmost/db/types/entity.types'; import { PaginationOptions } from '@docmost/db/pagination/pagination-options'; import { executeWithCursorPagination } from '@docmost/db/pagination/cursor-pagination'; -import { jsonObjectFrom } from 'kysely/helpers/postgres'; -import { ExpressionBuilder } from 'kysely'; +import { jsonArrayFrom, jsonObjectFrom } from 'kysely/helpers/postgres'; +import { ExpressionBuilder, sql } from 'kysely'; import { DB } from '@docmost/db/types/db'; @Injectable() @@ -25,6 +25,7 @@ export class PageHistoryRepo { 'icon', 'coverPhoto', 'lastUpdatedById', + 'contributorIds', 'spaceId', 'workspaceId', 'createdAt', @@ -44,6 +45,7 @@ export class PageHistoryRepo { .select(this.baseFields) .$if(opts?.includeContent, (qb) => qb.select('content')) .select((eb) => this.withLastUpdatedBy(eb)) + .select((eb) => this.withContributors(eb)) .where('id', '=', pageHistoryId) .executeTakeFirst(); } @@ -60,7 +62,10 @@ export class PageHistoryRepo { .executeTakeFirst(); } - async saveHistory(page: Page, trx?: KyselyTransaction): Promise { + async saveHistory( + page: Page, + opts?: { contributorIds?: string[]; trx?: KyselyTransaction }, + ): Promise { await this.insertPageHistory( { pageId: page.id, @@ -70,10 +75,11 @@ export class PageHistoryRepo { icon: page.icon, coverPhoto: page.coverPhoto, lastUpdatedById: page.lastUpdatedById ?? page.creatorId, + contributorIds: opts?.contributorIds, spaceId: page.spaceId, workspaceId: page.workspaceId, }, - trx, + opts?.trx, ); } @@ -82,6 +88,7 @@ export class PageHistoryRepo { .selectFrom('pageHistory') .select(this.baseFields) .select((eb) => this.withLastUpdatedBy(eb)) + .select((eb) => this.withContributors(eb)) .where('pageId', '=', pageId); return executeWithCursorPagination(query, { @@ -120,4 +127,17 @@ export class PageHistoryRepo { .whereRef('users.id', '=', 'pageHistory.lastUpdatedById'), ).as('lastUpdatedBy'); } + + withContributors(eb: ExpressionBuilder) { + return jsonArrayFrom( + eb + .selectFrom('users') + .select(['users.id', 'users.name', 'users.avatarUrl']) + .whereRef( + 'users.id', + '=', + sql`ANY(${eb.ref('pageHistory.contributorIds')})`, + ), + ).as('contributors'); + } } diff --git a/apps/server/src/database/types/db.d.ts b/apps/server/src/database/types/db.d.ts index a4197f4e..60b328ee 100644 --- a/apps/server/src/database/types/db.d.ts +++ b/apps/server/src/database/types/db.d.ts @@ -199,6 +199,7 @@ export interface GroupUsers { export interface PageHistory { content: Json | null; + contributorIds: Generated; coverPhoto: string | null; createdAt: Generated; icon: string | null; diff --git a/apps/server/src/integrations/queue/constants/queue.constants.ts b/apps/server/src/integrations/queue/constants/queue.constants.ts index 5c7aa29a..c194a28c 100644 --- a/apps/server/src/integrations/queue/constants/queue.constants.ts +++ b/apps/server/src/integrations/queue/constants/queue.constants.ts @@ -6,6 +6,7 @@ export enum QueueName { FILE_TASK_QUEUE = '{file-task-queue}', SEARCH_QUEUE = '{search-queue}', AI_QUEUE = '{ai-queue}', + HISTORY_QUEUE = '{history-queue}', } export enum QueueJob { @@ -58,4 +59,6 @@ export enum QueueJob { GENERATE_PAGE_EMBEDDINGS = 'generate-page-embeddings', DELETE_PAGE_EMBEDDINGS = 'delete-page-embeddings', + + PAGE_HISTORY = 'page-history', } diff --git a/apps/server/src/integrations/queue/constants/queue.interface.ts b/apps/server/src/integrations/queue/constants/queue.interface.ts index ce105f1c..cfcd7148 100644 --- a/apps/server/src/integrations/queue/constants/queue.interface.ts +++ b/apps/server/src/integrations/queue/constants/queue.interface.ts @@ -9,4 +9,8 @@ export interface IPageBacklinkJob { export interface IStripeSeatsSyncJob { workspaceId: string; +} + +export interface IPageHistoryJob { + pageId: string; } \ No newline at end of file diff --git a/apps/server/src/integrations/queue/queue.module.ts b/apps/server/src/integrations/queue/queue.module.ts index 6787e010..54037bce 100644 --- a/apps/server/src/integrations/queue/queue.module.ts +++ b/apps/server/src/integrations/queue/queue.module.ts @@ -73,6 +73,14 @@ import { BacklinksProcessor } from './processors/backlinks.processor'; attempts: 1, }, }), + BullModule.registerQueue({ + name: QueueName.HISTORY_QUEUE, + defaultJobOptions: { + removeOnComplete: true, + removeOnFail: true, + attempts: 2, + }, + }), ], exports: [BullModule], providers: [BacklinksProcessor],