fix interval

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 (
     <UnstyledButton
       p="xs"
@@ -39,25 +44,54 @@ const HistoryItem = memo(function HistoryItem({
       onMouseLeave={onHoverEnd}
       className={clsx(classes.history, { [classes.active]: isActive })}
     >
-      <Group wrap="nowrap">
+      <Text size="sm">{formattedDate(new Date(historyItem.createdAt))}</Text>
-        <div>
-          <Text size="sm">
-            {formattedDate(new Date(historyItem.createdAt))}
-          </Text>
 
-          <div style={{ flex: 1 }}>
+      <Group gap={6} wrap="nowrap" mt={4}>
-            <Group gap={4} wrap="nowrap">
+        {hasContributors ? (
-              <CustomAvatar
+          <>
-                size="sm"
+            <Tooltip.Group openDelay={300} closeDelay={100}>
-                avatarUrl={historyItem.lastUpdatedBy?.avatarUrl}
+              <Avatar.Group spacing={8}>
-                name={historyItem.lastUpdatedBy?.name}
+                {contributors.slice(0, MAX_VISIBLE_AVATARS).map((contributor) => (
-              />
+                  <Tooltip key={contributor.id} label={contributor.name} withArrow>
+                    <CustomAvatar
+                      size="sm"
+                      avatarUrl={contributor.avatarUrl}
+                      name={contributor.name}
+                    />
+                  </Tooltip>
+                ))}
+                {contributors.length > MAX_VISIBLE_AVATARS && (
+                  <Tooltip
+                    withArrow
+                    label={contributors.slice(MAX_VISIBLE_AVATARS).map((c) => (
+                      <div key={c.id}>{c.name}</div>
+                    ))}
+                  >
+                    <Avatar size="sm" color="gray">
+                      +{contributors.length - MAX_VISIBLE_AVATARS}
+                    </Avatar>
+                  </Tooltip>
+                )}
+              </Avatar.Group>
+            </Tooltip.Group>
+            {contributors.length === 1 && (
               <Text size="sm" c="dimmed" lineClamp={1}>
-                {historyItem.lastUpdatedBy?.name}
+                {contributors[0].name}
               </Text>
-            </Group>
+            )}
-          </div>
+          </>
-        </div>
+        ) : (
+          <>
+            <CustomAvatar
+              size="sm"
+              avatarUrl={historyItem.lastUpdatedBy?.avatarUrl}
+              name={historyItem.lastUpdatedBy?.name}
+            />
+            <Text size="sm" c="dimmed" lineClamp={1}>
+              {historyItem.lastUpdatedBy?.name}
+            </Text>
+          </>
+        )}
       </Group>
     </UnstyledButton>
   );

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) => ({
+      historyItems.map((item) => {
-        value: item.id,
+        const contributors = item.contributors;
-        label: formattedDate(new Date(item.createdAt)),
+        const hasContributors = contributors && contributors.length > 0;
-        userName: item.lastUpdatedBy?.name,
+        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],
   );
 

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[];
 }

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],

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;

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', {
+      await this.collabHistory.addContributors(pageId, editingUserIds);
-        page: {
-          ...page,
-          content: tiptapJson,
-          lastUpdatedById: context.user.id,
-        },
-      });
 
       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<void> {
+    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 },
+    );
+  }
 }

apps/server/src/collaboration/listeners/history.listener.ts

@@ -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);
-      }
-    }
-  }
-}

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<IPageHistoryJob, void>): Promise<void> {
+    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<void> {
+    if (this.worker) {
+      await this.worker.close();
+    }
+  }
+}

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<void> {
+    if (userIds.length === 0) return;
+    await this.redis.sadd(REDIS_KEY_PREFIX + pageId, ...userIds);
+  }
+
+  async popContributors(pageId: string): Promise<string[]> {
+    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<void> {
+    await this.redis.del(REDIS_KEY_PREFIX + pageId);
+  }
+}

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<any>): Promise<void> {
+  await db.schema
+    .alterTable('page_history')
+    .addColumn('contributor_ids', sql`uuid[]`, (col) => col.defaultTo('{}'))
+    .execute();
+}
+
+export async function down(db: Kysely<any>): Promise<void> {
+  await db.schema
+    .alterTable('page_history')
+    .dropColumn('contributor_ids')
+    .execute();
+}

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 { jsonArrayFrom, jsonObjectFrom } from 'kysely/helpers/postgres';
-import { ExpressionBuilder } from 'kysely';
+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<void> {
+  async saveHistory(
+    page: Page,
+    opts?: { contributorIds?: string[]; trx?: KyselyTransaction },
+  ): Promise<void> {
     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<DB, 'pageHistory'>) {
+    return jsonArrayFrom(
+      eb
+        .selectFrom('users')
+        .select(['users.id', 'users.name', 'users.avatarUrl'])
+        .whereRef(
+          'users.id',
+          '=',
+          sql`ANY(${eb.ref('pageHistory.contributorIds')})`,
+        ),
+    ).as('contributors');
+  }
 }

apps/server/src/database/types/db.d.ts

@@ -199,6 +199,7 @@ export interface GroupUsers {
 
 export interface PageHistory {
   content: Json | null;
+  contributorIds: Generated<string[] | null>;
   coverPhoto: string | null;
   createdAt: Generated<Timestamp>;
   icon: string | null;

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',
 }

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;
 }

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],