fix: WebSocket improvements (#2573)

apps/client/src/ee/oauth/pages/oauth-consent.tsx

@@ -36,6 +36,7 @@ import { currentUserAtom } from "@/features/user/atoms/current-user-atom";
 import { logout } from "@/features/auth/services/auth-service";
 import { ICurrentUser } from "@/features/user/types/user.types";
 import APP_ROUTE from "@/lib/app-route.ts";
+import { deleteLocalPageData } from "@/lib/local-page-data.ts";
 import {
   approveOAuthAuthorization,
   getOAuthAuthorizeInfo,
@@ -226,6 +227,7 @@ function ConsentCard({ info, currentUser, params }: ConsentCardProps) {
     }
     setCurrentUser(RESET);
     try {
+      await deleteLocalPageData();
       await logout();
     } finally {
       window.location.replace(

apps/client/src/features/auth/hooks/use-auth.ts

@@ -27,6 +27,7 @@ import APP_ROUTE, { getPostLoginRedirect } from "@/lib/app-route.ts";
 import { RESET } from "jotai/utils";
 import { useTranslation } from "react-i18next";
 import { isCloud } from "@/lib/config.ts";
+import { deleteLocalPageData } from "@/lib/local-page-data.ts";
 import { exchangeTokenRedirectUrl, getHostnameUrl } from "@/ee/utils.ts";
 
 export default function useAuth() {
@@ -165,6 +166,7 @@ export default function useAuth() {
 
   const handleLogout = async () => {
     setCurrentUser(RESET);
+    await deleteLocalPageData();
     await logout();
     window.location.replace(`${APP_ROUTE.AUTH.LOGIN}?logout=1`);
   };

apps/client/src/features/editor/full-editor.tsx

@@ -69,24 +69,24 @@ export function FullEditor({
   canComment,
 }: FullEditorProps) {
   const [user] = useAtom(userAtom);
-  const fullPageWidth = user.settings?.preferences?.fullPageWidth;
+  const fullPageWidth = user?.settings?.preferences?.fullPageWidth;
   const editorToolbarEnabled =
-    user.settings?.preferences?.editorToolbar ?? false;
+    user?.settings?.preferences?.editorToolbar ?? false;
   const [currentPageEditMode, setCurrentPageEditMode] = useAtom(
     currentPageEditModeAtom,
   );
   const userPageEditMode =
-    user.settings?.preferences?.pageEditMode ?? PageEditMode.Edit;
+    user?.settings?.preferences?.pageEditMode ?? PageEditMode.Edit;
   const isEditMode = currentPageEditMode === PageEditMode.Edit;
 
   // Apply the user's saved preference only once on initial load, not on every
   // page navigation — so the mode sticks across navigations within a session.
   useEffect(() => {
-    if (!defaultEditModeApplied) {
+    if (user && !defaultEditModeApplied) {
       setCurrentPageEditMode(userPageEditMode as PageEditMode);
       defaultEditModeApplied = true;
     }
-  }, [userPageEditMode, setCurrentPageEditMode]);
+  }, [user, userPageEditMode, setCurrentPageEditMode]);
 
   return (
     <Container

apps/client/src/features/editor/page-editor.tsx

@@ -119,6 +119,31 @@ export default function PageEditor({
   const handleStateless = ({ payload }: onStatelessParameters) => {
     try {
       const message = JSON.parse(payload);
+      if (message?.type === "page.permissions") {
+        // applied locally, a refetch can still read the old access from the server cache
+        for (const key of [slugId, pageId]) {
+          queryClient.setQueryData<IPage>(
+            ["pages", key],
+            (page) =>
+              page && {
+                ...page,
+                permissions: { ...page.permissions, canEdit: false },
+              },
+          );
+        }
+        if (message.access === "none") {
+          queryClient
+            .getQueryCache()
+            .find({ queryKey: ["pages", slugId], exact: true })
+            ?.setState({
+              status: "error",
+              fetchStatus: "idle",
+              data: undefined,
+              error: Object.assign(new Error("Forbidden"), { status: 403 }),
+            });
+        }
+        return;
+      }
       if (message?.type !== "page.updated" || !message.updatedAt) return;
       const pageData = queryClient.getQueryData<IPage>(["pages", slugId]);
       if (pageData) {

apps/client/src/lib/local-page-data.test.ts

@@ -0,0 +1,35 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { deleteLocalPageData } from "./local-page-data";
+
+function stubIndexedDB(names: string[]) {
+  const deleteDatabase = vi.fn(() => {
+    const request: Record<string, () => void> = {};
+    setTimeout(() => request.onsuccess?.());
+    return request;
+  });
+  vi.stubGlobal("indexedDB", {
+    databases: async () => names.map((name) => ({ name, version: 1 })),
+    deleteDatabase,
+  });
+  return deleteDatabase;
+}
+
+describe("deleteLocalPageData", () => {
+  afterEach(() => {
+    vi.unstubAllGlobals();
+  });
+
+  it("deletes only the page databases", async () => {
+    const deleteDatabase = stubIndexedDB(["page.a", "page.b", "other-db"]);
+
+    await deleteLocalPageData();
+
+    expect(deleteDatabase.mock.calls).toEqual([["page.a"], ["page.b"]]);
+  });
+
+  it("does nothing when the browser cannot list databases", async () => {
+    vi.stubGlobal("indexedDB", { deleteDatabase: vi.fn() });
+
+    await expect(deleteLocalPageData()).resolves.toBeUndefined();
+  });
+});

apps/client/src/lib/local-page-data.ts

@@ -0,0 +1,25 @@
+// y-indexeddb names each database after the collab document, page.<pageId>
+const PAGE_DATABASE_PREFIX = "page.";
+
+export async function deleteLocalPageData(): Promise<void> {
+  try {
+    const databases = (await indexedDB.databases?.()) ?? [];
+    const pageDatabaseNames = databases
+      .map((database) => database.name)
+      .filter((name) => name?.startsWith(PAGE_DATABASE_PREFIX));
+
+    await Promise.all(
+      pageDatabaseNames.map(
+        (name) =>
+          new Promise<void>((resolve) => {
+            const request = indexedDB.deleteDatabase(name);
+            request.onsuccess = () => resolve();
+            request.onerror = () => resolve();
+            request.onblocked = () => resolve();
+          }),
+      ),
+    );
+  } catch {
+    // local cleanup must never block logout
+  }
+}

apps/server/src/app.module.ts

@@ -7,6 +7,7 @@ import { AuditActorInterceptor } from './common/interceptors/audit-actor.interce
 import { CoreModule } from './core/core.module';
 import { EnvironmentModule } from './integrations/environment/environment.module';
 import { CollaborationModule } from './collaboration/collaboration.module';
+import { CollabAccessModule } from './collaboration/collab-access.module';
 import { WsModule } from './ws/ws.module';
 import { DatabaseModule } from '@docmost/db/database.module';
 import { StorageModule } from './integrations/storage/storage.module';
@@ -83,6 +84,7 @@ try {
       inject: [EnvironmentService],
     }),
     CollaborationModule,
+    CollabAccessModule,
     WsModule,
     QueueModule,
     StaticModule,

apps/server/src/collaboration/collab-access.module.ts

@@ -0,0 +1,9 @@
+import { Global, Module } from '@nestjs/common';
+import { CollabAccessService } from './services/collab-access.service';
+
+@Global()
+@Module({
+  providers: [CollabAccessService],
+  exports: [CollabAccessService],
+})
+export class CollabAccessModule {}

apps/server/src/collaboration/constants.ts

@@ -1,3 +1,4 @@
 export const HISTORY_INTERVAL = 5 * 60 * 1000;
 export const HISTORY_FAST_INTERVAL = 60 * 1000;
 export const HISTORY_FAST_THRESHOLD = 5 * 60 * 1000;
+export const COLLAB_ACCESS_RECHECK_CHANNEL = 'collab:access-recheck';

apps/server/src/collaboration/extensions/authentication.extension.spec.ts

@@ -0,0 +1,221 @@
+import { Forbidden } from '@hocuspocus/common';
+import { NotFoundException, UnauthorizedException } from '@nestjs/common';
+import { AuthenticationExtension } from './authentication.extension';
+import { COLLAB_ACCESS_RECHECK_CHANNEL } from '../constants';
+
+const WRITER = [{ userId: 'any', role: 'writer' }];
+const READER = [{ userId: 'any', role: 'reader' }];
+const NO_ACCESS = JSON.stringify({ type: 'page.permissions', access: 'none' });
+const READ_ACCESS = JSON.stringify({ type: 'page.permissions', access: 'read' });
+
+function makeConnection(userId: string) {
+  return {
+    context: { user: { id: userId, workspaceId: 'ws_1' } },
+    readOnly: false,
+    close: jest.fn(),
+    sendStateless: jest.fn(),
+  };
+}
+
+async function makeExtension(
+  spaceRoles: Record<string, unknown[] | undefined>,
+  documents: Record<string, ReturnType<typeof makeConnection>[]> = {},
+) {
+  const subscriber = {
+    on: jest.fn(),
+    subscribe: jest.fn(async () => undefined),
+    disconnect: jest.fn(),
+  };
+  const spaceMemberRepo = {
+    getUserSpaceRoles: jest.fn(async (userId: string) => spaceRoles[userId]),
+  };
+  const pagePermissionRepo = {
+    canUserEditPage: jest.fn(async () => ({
+      hasAnyRestriction: false,
+      canAccess: true,
+      canEdit: true,
+    })),
+  };
+  const extension = new AuthenticationExtension(
+    { verifyJwt: jest.fn(async (token: string) => ({ sub: token, workspaceId: 'ws_1' })) } as any,
+    {
+      findById: jest.fn(async (id: string) => {
+        if (id === 'user_broken') throw new Error('db down');
+        return { id, workspaceId: 'ws_1' };
+      }),
+    } as any,
+    {
+      findById: jest.fn(async (id: string) => ({
+        id,
+        spaceId: 'space_1',
+        workspaceId: id === 'page_other_ws' ? 'ws_2' : 'ws_1',
+        deletedAt: null,
+      })),
+      getPageAndDescendants: jest.fn(async (id: string) => [{ id }, { id: 'child_page' }]),
+    } as any,
+    spaceMemberRepo as any,
+    pagePermissionRepo as any,
+    { getOrThrow: () => ({ duplicate: () => subscriber }) } as any,
+  );
+  const instance = {
+    documents: new Map(
+      Object.entries(documents).map(([name, connections]) => [
+        name,
+        { name, getConnections: () => connections },
+      ]),
+    ),
+  };
+  await extension.onConfigure({ instance } as any);
+  return { extension, subscriber, spaceMemberRepo, pagePermissionRepo };
+}
+
+describe('AuthenticationExtension access recheck', () => {
+  it('closes connections of users who lost access and leaves others alone', async () => {
+    const removed = makeConnection('user_removed');
+    const other = makeConnection('user_other');
+    const { extension } = await makeExtension(
+      { user_removed: undefined, user_other: WRITER },
+      { 'page.page_1': [removed, other] },
+    );
+
+    await extension.recheckAccess({ userIds: ['user_removed'] });
+
+    expect(removed.sendStateless).toHaveBeenCalledWith(NO_ACCESS);
+    expect(removed.readOnly).toBe(true);
+    expect(removed.close).toHaveBeenCalledWith(Forbidden);
+    expect(other.sendStateless).not.toHaveBeenCalled();
+    expect(other.close).not.toHaveBeenCalled();
+  });
+
+  it('makes downgraded editors read-only without closing them', async () => {
+    const downgraded = makeConnection('user_reader');
+    const { extension } = await makeExtension(
+      { user_reader: READER },
+      { 'page.page_1': [downgraded] },
+    );
+
+    await extension.recheckAccess({ userIds: ['user_reader'] });
+
+    expect(downgraded.readOnly).toBe(true);
+    expect(downgraded.sendStateless).toHaveBeenCalledWith(READ_ACCESS);
+    expect(downgraded.close).not.toHaveBeenCalled();
+  });
+
+  it('rechecks connections on pages under a changed page only', async () => {
+    const underChangedPage = makeConnection('user_1');
+    const elsewhere = makeConnection('user_1');
+    const { extension, spaceMemberRepo } = await makeExtension(
+      { user_1: undefined },
+      { 'page.child_page': [underChangedPage], 'page.other_page': [elsewhere] },
+    );
+
+    await extension.recheckAccess({ pageIds: ['page_1'] });
+
+    expect(underChangedPage.close).toHaveBeenCalledWith(Forbidden);
+    expect(elsewhere.close).not.toHaveBeenCalled();
+    expect(spaceMemberRepo.getUserSpaceRoles).toHaveBeenCalledWith(
+      'user_1',
+      'space_1',
+      { skipCache: true },
+    );
+  });
+
+  it('rechecks when an access message arrives on the channel', async () => {
+    const removed = makeConnection('user_removed');
+    const { subscriber } = await makeExtension(
+      { user_removed: undefined },
+      { 'page.page_1': [removed] },
+    );
+    const handler = (name: string) =>
+      subscriber.on.mock.calls.find(([event]) => event === name)[1];
+
+    expect(subscriber.on).toHaveBeenCalledWith('error', expect.any(Function));
+    handler('ready')();
+    handler('ready')();
+    expect(subscriber.subscribe).toHaveBeenCalledTimes(2);
+    expect(subscriber.subscribe).toHaveBeenCalledWith(
+      COLLAB_ACCESS_RECHECK_CHANNEL,
+    );
+    const onMessage = handler('message');
+    await onMessage(
+      COLLAB_ACCESS_RECHECK_CHANNEL,
+      JSON.stringify({ userIds: ['user_removed'] }),
+    );
+
+    expect(removed.close).toHaveBeenCalledWith(Forbidden);
+  });
+
+  it('keeps rechecking after one connection fails', async () => {
+    const broken = makeConnection('user_broken');
+    const removed = makeConnection('user_removed');
+    const { extension } = await makeExtension(
+      { user_removed: undefined },
+      { 'page.page_1': [broken, removed] },
+    );
+
+    await extension.recheckAccess({ userIds: ['user_broken', 'user_removed'] });
+
+    expect(broken.close).not.toHaveBeenCalled();
+    expect(removed.close).toHaveBeenCalledWith(Forbidden);
+  });
+});
+
+describe('AuthenticationExtension onAuthenticate', () => {
+  it('checks permissions without the cache', async () => {
+    const { extension, spaceMemberRepo, pagePermissionRepo } =
+      await makeExtension({ user_reader: READER });
+
+    await extension.onAuthenticate({
+      documentName: 'page.page_1',
+      token: 'user_reader',
+      connectionConfig: { readOnly: false },
+    } as any);
+
+    expect(spaceMemberRepo.getUserSpaceRoles).toHaveBeenCalledWith(
+      'user_reader',
+      'space_1',
+      { skipCache: true },
+    );
+    expect(pagePermissionRepo.canUserEditPage).toHaveBeenCalledWith(
+      'user_reader',
+      'page_1',
+      { skipCache: true },
+    );
+  });
+
+  it('opens readers read-only and rejects users without access', async () => {
+    const { extension } = await makeExtension({
+      user_reader: READER,
+      user_none: undefined,
+    });
+    const payload = (token: string) =>
+      ({
+        documentName: 'page.page_1',
+        token,
+        connectionConfig: { readOnly: false },
+      }) as any;
+
+    const readerPayload = payload('user_reader');
+    await extension.onAuthenticate(readerPayload);
+    expect(readerPayload.connectionConfig.readOnly).toBe(true);
+
+    await expect(extension.onAuthenticate(payload('user_none'))).rejects.toThrow(
+      UnauthorizedException,
+    );
+  });
+
+  it('rejects pages from another workspace', async () => {
+    const { extension, spaceMemberRepo } = await makeExtension({
+      user_writer: WRITER,
+    });
+
+    await expect(
+      extension.onAuthenticate({
+        documentName: 'page.page_other_ws',
+        token: 'user_writer',
+        connectionConfig: { readOnly: false },
+      } as any),
+    ).rejects.toThrow(NotFoundException);
+    expect(spaceMemberRepo.getUserSpaceRoles).not.toHaveBeenCalled();
+  });
+});

apps/server/src/collaboration/extensions/authentication.extension.ts

@@ -1,24 +1,40 @@
-import { Extension, onAuthenticatePayload } from '@hocuspocus/server';
+import {
+  Connection,
+  Extension,
+  Hocuspocus,
+  onAuthenticatePayload,
+  onConfigurePayload,
+} from '@hocuspocus/server';
+import { Forbidden } from '@hocuspocus/common';
 import {
   Injectable,
   Logger,
   NotFoundException,
   UnauthorizedException,
 } from '@nestjs/common';
+import { RedisService } from '@nestjs-labs/nestjs-ioredis';
+import type { Redis } from 'ioredis';
 import { TokenService } from '../../core/auth/services/token.service';
 import { UserRepo } from '@docmost/db/repos/user/user.repo';
 import { PageRepo } from '@docmost/db/repos/page/page.repo';
 import { SpaceMemberRepo } from '@docmost/db/repos/space/space-member.repo';
 import { PagePermissionRepo } from '@docmost/db/repos/page/page-permission.repo';
 import { findHighestUserSpaceRole } from '@docmost/db/repos/space/utils';
+import { Page, User } from '@docmost/db/types/entity.types';
 import { SpaceRole } from '../../common/helpers/types/permission';
 import { isUserDisabled } from '../../common/helpers';
 import { getPageId } from '../collaboration.util';
+import { COLLAB_ACCESS_RECHECK_CHANNEL } from '../constants';
+import { CollabAccessRecheck } from '../services/collab-access.service';
 import { JwtCollabPayload, JwtType } from '../../core/auth/dto/jwt-payload';
 
+type CollabAccess = 'none' | 'read' | 'write';
+
 @Injectable()
 export class AuthenticationExtension implements Extension {
   private readonly logger = new Logger(AuthenticationExtension.name);
+  private instance: Hocuspocus;
+  private subscriber: Redis;
 
   constructor(
     private tokenService: TokenService,
@@ -26,8 +42,34 @@ export class AuthenticationExtension implements Extension {
     private pageRepo: PageRepo,
     private readonly spaceMemberRepo: SpaceMemberRepo,
     private readonly pagePermissionRepo: PagePermissionRepo,
+    private readonly redisService: RedisService,
   ) {}
 
+  async onConfigure({ instance }: onConfigurePayload) {
+    this.instance = instance;
+    this.subscriber = this.redisService.getOrThrow().duplicate();
+    this.subscriber.on('error', () => {});
+    this.subscriber.on('message', async (_channel, message) => {
+      try {
+        await this.recheckAccess(JSON.parse(message));
+      } catch (err) {
+        this.logger.error({ err }, 'Failed to recheck collab access');
+      }
+    });
+    // resubscribe on every ready, ioredis never replays a SUBSCRIBE that failed while redis was down
+    this.subscriber.on('ready', () => {
+      this.subscriber
+        .subscribe(COLLAB_ACCESS_RECHECK_CHANNEL)
+        .catch((err) =>
+          this.logger.error({ err }, 'Failed to subscribe to collab rechecks'),
+        );
+    });
+  }
+
+  async onDestroy() {
+    this.subscriber?.disconnect();
+  }
+
   async onAuthenticate(data: onAuthenticatePayload) {
     const { documentName, token } = data;
     const pageId = getPageId(documentName);
@@ -43,61 +85,28 @@ export class AuthenticationExtension implements Extension {
     const userId = jwtPayload.sub;
     const workspaceId = jwtPayload.workspaceId;
 
-    const user = await this.userRepo.findById(userId, workspaceId);
+    const [user, page] = await Promise.all([
+      this.userRepo.findById(userId, workspaceId),
+      this.pageRepo.findById(pageId),
+    ]);
 
-    if (!user) {
+    if (!user || isUserDisabled(user)) {
       throw new UnauthorizedException();
     }
 
-    if (isUserDisabled(user)) {
-      throw new UnauthorizedException();
-    }
-
-    const page = await this.pageRepo.findById(pageId);
-    if (!page) {
+    if (!page || page.workspaceId !== workspaceId) {
       this.logger.debug(`Page not found: ${pageId}`);
       throw new NotFoundException('Page not found');
     }
 
-    const userSpaceRoles = await this.spaceMemberRepo.getUserSpaceRoles(
-      user.id,
-      page.spaceId,
-    );
+    const access = await this.getAccess(user, page);
 
-    const userSpaceRole = findHighestUserSpaceRole(userSpaceRoles);
-
-    if (!userSpaceRole) {
-      this.logger.warn(`User not authorized to access page: ${pageId}`);
+    if (access === 'none') {
+      this.logger.warn(`User ${user.id} not authorized to access page: ${pageId}`);
       throw new UnauthorizedException();
     }
 
-    // Check page-level permissions
-    const { hasAnyRestriction, canAccess, canEdit } =
-      await this.pagePermissionRepo.canUserEditPage(user.id, page.id);
-
-    if (hasAnyRestriction) {
-      if (!canAccess) {
-        this.logger.warn(
-          `User ${user.id} denied page-level access to page: ${pageId}`,
-        );
-        throw new UnauthorizedException();
-      }
-
-      if (!canEdit) {
-        data.connectionConfig.readOnly = true;
-        this.logger.debug(
-          `User ${user.id} granted readonly access to restricted page: ${pageId}`,
-        );
-      }
-    } else {
-      // No restrictions - use space-level permissions
-      if (userSpaceRole === SpaceRole.READER) {
-        data.connectionConfig.readOnly = true;
-        this.logger.debug(`User granted readonly access to page: ${pageId}`);
-      }
-    }
-
-    if (page.deletedAt) {
+    if (access === 'read') {
       data.connectionConfig.readOnly = true;
     }
 
@@ -107,4 +116,92 @@ export class AuthenticationExtension implements Extension {
       user,
     };
   }
+
+  async recheckAccess({
+    userIds = [],
+    pageIds = [],
+  }: CollabAccessRecheck): Promise<void> {
+    if (!this.instance?.documents.size) return;
+
+    const recheckUserIds = new Set(userIds);
+    const recheckPageIds = new Set<string>();
+    for (const pageId of pageIds) {
+      const pages = await this.pageRepo.getPageAndDescendants(pageId, {
+        includeContent: false,
+      });
+      pages.forEach((page) => recheckPageIds.add(page.id));
+    }
+
+    for (const document of this.instance.documents.values()) {
+      const pageId = getPageId(document.name);
+      const connections = document.getConnections().filter((connection) => {
+        const userId = connection.context?.user?.id;
+        return (
+          userId && (recheckPageIds.has(pageId) || recheckUserIds.has(userId))
+        );
+      });
+      if (connections.length === 0) continue;
+
+      for (const connection of connections) {
+        try {
+          await this.recheckConnection(connection, pageId);
+        } catch (err) {
+          this.logger.error(
+            { err },
+            `Failed to recheck collab access on ${document.name}`,
+          );
+        }
+      }
+    }
+  }
+
+  private async recheckConnection(
+    connection: Connection,
+    pageId: string,
+  ): Promise<void> {
+    const { id, workspaceId } = connection.context.user;
+    const [user, page] = await Promise.all([
+      this.userRepo.findById(id, workspaceId),
+      this.pageRepo.findById(pageId),
+    ]);
+
+    let access: CollabAccess = 'none';
+    if (user && !isUserDisabled(user) && page) {
+      access = await this.getAccess(user, page);
+    }
+
+    if (access === 'write' || (access === 'read' && connection.readOnly)) {
+      return;
+    }
+
+    connection.sendStateless(
+      JSON.stringify({ type: 'page.permissions', access }),
+    );
+    connection.readOnly = true;
+    if (access === 'none') {
+      connection.close(Forbidden);
+    }
+  }
+
+  // a connection outlives this check, so it must not see a stale cached allow
+  private async getAccess(user: User, page: Page): Promise<CollabAccess> {
+    const [spaceRoles, { hasAnyRestriction, canAccess, canEdit }] =
+      await Promise.all([
+        this.spaceMemberRepo.getUserSpaceRoles(user.id, page.spaceId, {
+          skipCache: true,
+        }),
+        this.pagePermissionRepo.canUserEditPage(user.id, page.id, {
+          skipCache: true,
+        }),
+      ]);
+
+    const userSpaceRole = findHighestUserSpaceRole(spaceRoles);
+    if (!userSpaceRole) return 'none';
+    if (hasAnyRestriction && !canAccess) return 'none';
+
+    const canWrite = hasAnyRestriction
+      ? canEdit
+      : userSpaceRole !== SpaceRole.READER;
+    return canWrite && !page.deletedAt ? 'write' : 'read';
+  }
 }

apps/server/src/collaboration/extensions/persistence.extension.ts

@@ -138,8 +138,6 @@ export class PersistenceExtension implements Extension {
           return;
         }
 
-        await this.collabHistory.addContributors(pageId, editingUserIds);
-
         let contributorIds = undefined;
         try {
           const existingContributors = page.contributorIds || [];
@@ -177,6 +175,15 @@ export class PersistenceExtension implements Extension {
     }
 
     if (page) {
+      try {
+        await this.collabHistory.addContributors(pageId, editingUserIds);
+      } catch (err) {
+        this.logger.error(
+          { err, pageId },
+          'Failed to add history contributors for page',
+        );
+      }
+
       document.broadcastStateless(
         JSON.stringify({
           type: 'page.updated',

apps/server/src/collaboration/services/collab-access.service.spec.ts

@@ -0,0 +1,29 @@
+import { CollabAccessService } from './collab-access.service';
+import { COLLAB_ACCESS_RECHECK_CHANNEL } from '../constants';
+
+function makeService(publish: (...args: any[]) => Promise<number>) {
+  return new CollabAccessService({ getOrThrow: () => ({ publish }) } as any);
+}
+
+describe('CollabAccessService', () => {
+  it('publishes the users or pages to recheck', async () => {
+    const publish = jest.fn(async () => 1);
+
+    await makeService(publish).recheckAccess({ userIds: ['user_1'] });
+    await makeService(publish).recheckAccess({ userIds: [], pageIds: [] });
+
+    expect(publish.mock.calls).toEqual([
+      [COLLAB_ACCESS_RECHECK_CHANNEL, JSON.stringify({ userIds: ['user_1'] })],
+    ]);
+  });
+
+  it('does not throw when publishing fails', async () => {
+    const service = makeService(async () => {
+      throw new Error('redis down');
+    });
+
+    await expect(
+      service.recheckAccess({ userIds: ['user_1'] }),
+    ).resolves.toBeUndefined();
+  });
+});

apps/server/src/collaboration/services/collab-access.service.ts

@@ -0,0 +1,32 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { RedisService } from '@nestjs-labs/nestjs-ioredis';
+import { COLLAB_ACCESS_RECHECK_CHANNEL } from '../constants';
+
+export type CollabAccessRecheck = {
+  userIds?: string[];
+  pageIds?: string[];
+};
+
+@Injectable()
+export class CollabAccessService {
+  private readonly logger = new Logger(CollabAccessService.name);
+
+  constructor(private readonly redisService: RedisService) {}
+
+  async recheckAccess(recheck: CollabAccessRecheck): Promise<void> {
+    if (!recheck.userIds?.length && !recheck.pageIds?.length) return;
+    // not awaited, ioredis queues the command and can hold the reply for minutes while redis reconnects
+    try {
+      this.redisService
+        .getOrThrow()
+        .publish(COLLAB_ACCESS_RECHECK_CHANNEL, JSON.stringify(recheck))
+        .catch((err) => this.logError(err));
+    } catch (err) {
+      this.logError(err);
+    }
+  }
+
+  private logError(err: unknown): void {
+    this.logger.error({ err }, 'Failed to publish collab access recheck');
+  }
+}

apps/server/src/common/helpers/with-cache.spec.ts

@@ -0,0 +1,29 @@
+import { withCache } from './with-cache';
+
+function makeCache(cached?: unknown) {
+  return {
+    get: jest.fn(async () => (cached === undefined ? undefined : { v: cached })),
+    set: jest.fn(async () => undefined),
+  };
+}
+
+describe('withCache', () => {
+  it('returns a cached value without running the lookup', async () => {
+    const cache = makeCache('cached');
+    const lookup = jest.fn(async () => 'fresh');
+
+    await expect(withCache(cache as any, 'key', 1000, lookup)).resolves.toBe('cached');
+    expect(lookup).not.toHaveBeenCalled();
+  });
+
+  it('skips the cache entirely when asked', async () => {
+    const cache = makeCache('cached');
+    const lookup = jest.fn(async () => 'fresh');
+
+    await expect(
+      withCache(cache as any, 'key', 1000, lookup, { skipCache: true }),
+    ).resolves.toBe('fresh');
+    expect(cache.get).not.toHaveBeenCalled();
+    expect(cache.set).not.toHaveBeenCalled();
+  });
+});

apps/server/src/common/helpers/with-cache.ts

@@ -1,11 +1,16 @@
 import { Cache } from 'cache-manager';
 
+export type CacheOptions = { skipCache?: boolean };
+
 export async function withCache<T>(
   cacheManager: Cache,
   key: string,
   ttlMs: number,
   fn: () => Promise<T>,
+  opts?: CacheOptions,
 ): Promise<T> {
+  if (opts?.skipCache) return fn();
+
   try {
     const cached = await cacheManager.get<{ v: T }>(key);
     if (cached !== undefined && cached !== null) {
@@ -17,11 +22,9 @@ export async function withCache<T>(
 
   const value = await fn();
 
-  try {
-    await cacheManager.set(key, { v: value }, ttlMs);
-  } catch (err) {
-    console.warn(`[withCache] set failed for "${key}"`, err);
-  }
+  cacheManager
+    .set(key, { v: value }, ttlMs)
+    .catch((err) => console.warn(`[withCache] set failed for "${key}"`, err));
 
   return value;
 }

apps/server/src/core/auth/services/auth.service.ts

@@ -33,6 +33,7 @@ import { UserTokenType } from '../auth.constants';
 import { KyselyDB } from '@docmost/db/types/kysely.types';
 import { InjectKysely } from 'nestjs-kysely';
 import { executeTx } from '@docmost/db/utils';
+import { WsService } from '../../../ws/ws.service';
 import { VerifyUserTokenDto } from '../dto/verify-user-token.dto';
 import { DomainService } from '../../../integrations/environment/domain.service';
 import { AuditEvent, AuditResource } from '../../../common/events/audit-events';
@@ -61,6 +62,7 @@ export class AuthService {
     private eventEmitter: EventEmitter2,
     @InjectKysely() private readonly db: KyselyDB,
     @Inject(AUDIT_SERVICE) private readonly auditService: IAuditService,
+    private wsService: WsService,
   ) {}
 
   async login(loginDto: LoginDto, workspaceId: string) {
@@ -155,8 +157,10 @@ export class AuthService {
         userId,
         workspaceId,
       );
+      this.wsService.disconnectOtherSessions(userId, currentSessionId);
     } else {
       await this.userSessionRepo.deleteByUserId(userId, workspaceId);
+      this.wsService.disconnectUser(userId);
     }
 
     this.auditService.log({
@@ -273,6 +277,7 @@ export class AuthService {
     });
 
     await this.userSessionRepo.deleteByUserId(user.id, workspace.id);
+    this.wsService.disconnectUser(user.id);
 
     // A failed revocation must not block the reset itself; log loudly instead.
     try {

apps/server/src/core/casl/abilities/space-ability.factory.ts

@@ -13,14 +13,16 @@ import {
   SpaceCaslSubject,
 } from '../interfaces/space-ability.type';
 import { findHighestUserSpaceRole } from '@docmost/db/repos/space/utils';
+import { CacheOptions } from '../../../common/helpers/with-cache';
 
 @Injectable()
 export default class SpaceAbilityFactory {
   constructor(private readonly spaceMemberRepo: SpaceMemberRepo) {}
-  async createForUser(user: User, spaceId: string) {
+  async createForUser(user: User, spaceId: string, opts?: CacheOptions) {
     const userSpaceRoles = await this.spaceMemberRepo.getUserSpaceRoles(
       user.id,
       spaceId,
+      opts,
     );
 
     const userSpaceRole = findHighestUserSpaceRole(userSpaceRoles);

apps/server/src/core/group/services/group-user.service.ts

@@ -21,6 +21,11 @@ import {
   IAuditService,
 } from '../../../integrations/audit/audit.service';
 import { dbOrTx } from '@docmost/db/utils';
+import { WsService } from '../../../ws/ws.service';
+import { PagePermissionRepo } from '@docmost/db/repos/page/page-permission.repo';
+import { CollabAccessService } from '../../../collaboration/services/collab-access.service';
+
+export type GroupAccess = { spaceIds: string[]; pageIds: string[] };
 
 @Injectable()
 export class GroupUserService {
@@ -34,6 +39,9 @@ export class GroupUserService {
     private readonly favoriteRepo: FavoriteRepo,
     @InjectKysely() private readonly db: KyselyDB,
     @Inject(AUDIT_SERVICE) private readonly auditService: IAuditService,
+    private readonly wsService: WsService,
+    private readonly pagePermissionRepo: PagePermissionRepo,
+    private readonly collabAccessService: CollabAccessService,
   ) {}
 
   async getGroupUsers(
@@ -134,27 +142,20 @@ export class GroupUserService {
       throw new BadRequestException('Group member not found');
     }
 
-    const spaceIds = await this.spaceMemberRepo.getSpaceIdsByGroupId(groupId);
+    const groupAccess = await this.getGroupAccess([groupId]);
 
     // TODO: use queue instead
     await executeTx(this.db, async (trx) => {
       await this.groupUserRepo.delete(userId, groupId, { trx });
-
-      for (const spaceId of spaceIds) {
-        await this.watcherRepo.deleteByUsersWithoutSpaceAccess(
-          [userId],
-          spaceId,
-          { trx },
-        );
-
-        await this.favoriteRepo.deleteByUsersWithoutSpaceAccess(
-          [userId],
-          spaceId,
-          { trx },
-        );
-      }
+      await this.removeWatchersAndFavoritesWithoutAccess(
+        [userId],
+        groupAccess.spaceIds,
+        trx,
+      );
     });
 
+    await this.revokeRealtimeAccess([userId], groupAccess);
+
     this.auditService.log({
       event: AuditEvent.GROUP_MEMBER_REMOVED,
       resourceType: AuditResource.GROUP,
@@ -170,4 +171,57 @@ export class GroupUserService {
       },
     });
   }
+
+  // read before deleting a group, its page permissions cascade away with it
+  async getGroupAccess(
+    groupIds: string[],
+    opts?: { trx?: KyselyTransaction },
+  ): Promise<GroupAccess> {
+    const spaceIds = await this.spaceMemberRepo.getSpaceIdsByGroupIds(
+      groupIds,
+      opts,
+    );
+    const pageIds = await this.pagePermissionRepo.getPageIdsByGroupIds(
+      groupIds,
+      opts,
+    );
+    return { spaceIds, pageIds };
+  }
+
+  async removeWatchersAndFavoritesWithoutAccess(
+    userIds: string[],
+    spaceIds: string[],
+    trx: KyselyTransaction,
+  ): Promise<void> {
+    for (const spaceId of spaceIds) {
+      await this.watcherRepo.deleteByUsersWithoutSpaceAccess(
+        userIds,
+        spaceId,
+        { trx },
+      );
+      await this.favoriteRepo.deleteByUsersWithoutSpaceAccess(
+        userIds,
+        spaceId,
+        { trx },
+      );
+    }
+  }
+
+  // call after commit, it never throws
+  async revokeRealtimeAccess(
+    userIds: string[],
+    groupAccess: GroupAccess,
+  ): Promise<void> {
+    if (userIds.length === 0) return;
+    if (!groupAccess.spaceIds.length && !groupAccess.pageIds.length) return;
+
+    await this.collabAccessService.recheckAccess({ userIds });
+    await this.wsService.removeUsersWithoutSpaceAccess(
+      userIds,
+      groupAccess.spaceIds,
+    );
+    await this.wsService.removeBaseSubscribersWithoutAccess(
+      groupAccess.pageIds,
+    );
+  }
 }

apps/server/src/core/group/services/group.service.ts

@@ -11,12 +11,9 @@ import { UpdateGroupDto } from '../dto/update-group.dto';
 import { KyselyDB, KyselyTransaction } from '@docmost/db/types/kysely.types';
 import { GroupRepo } from '@docmost/db/repos/group/group.repo';
 import { GroupUserRepo } from '@docmost/db/repos/group/group-user.repo';
-import { SpaceMemberRepo } from '@docmost/db/repos/space/space-member.repo';
 import { Group, InsertableGroup, User } from '@docmost/db/types/entity.types';
 import { CursorPaginationResult } from '@docmost/db/pagination/cursor-pagination';
 import { GroupUserService } from './group-user.service';
-import { WatcherRepo } from '@docmost/db/repos/watcher/watcher.repo';
-import { FavoriteRepo } from '@docmost/db/repos/favorite/favorite.repo';
 import { executeTx } from '@docmost/db/utils';
 import { InjectKysely } from 'nestjs-kysely';
 import { AuditEvent, AuditResource } from '../../../common/events/audit-events';
@@ -31,11 +28,8 @@ export class GroupService {
   constructor(
     private groupRepo: GroupRepo,
     private groupUserRepo: GroupUserRepo,
-    private spaceMemberRepo: SpaceMemberRepo,
     @Inject(forwardRef(() => GroupUserService))
     private groupUserService: GroupUserService,
-    private readonly watcherRepo: WatcherRepo,
-    private readonly favoriteRepo: FavoriteRepo,
     @InjectKysely() private readonly db: KyselyDB,
     @Inject(AUDIT_SERVICE) private readonly auditService: IAuditService,
   ) {}
@@ -176,30 +170,23 @@ export class GroupService {
       throw new BadRequestException('You cannot delete a default group');
     }
 
-    const [userIds, spaceIds] = await Promise.all([
+    const [userIds, groupAccess] = await Promise.all([
       this.groupUserRepo.getUserIdsByGroupId(groupId),
-      this.spaceMemberRepo.getSpaceIdsByGroupId(groupId),
+      this.groupUserService.getGroupAccess([groupId]),
     ]);
 
     // TODO: use queue instead
     await executeTx(this.db, async (trx) => {
       await this.groupRepo.delete(groupId, workspaceId, { trx });
-
-      for (const spaceId of spaceIds) {
-        await this.watcherRepo.deleteByUsersWithoutSpaceAccess(
-          userIds,
-          spaceId,
-          { trx },
-        );
-
-        await this.favoriteRepo.deleteByUsersWithoutSpaceAccess(
-          userIds,
-          spaceId,
-          { trx },
-        );
-      }
+      await this.groupUserService.removeWatchersAndFavoritesWithoutAccess(
+        userIds,
+        groupAccess.spaceIds,
+        trx,
+      );
     });
 
+    await this.groupUserService.revokeRealtimeAccess(userIds, groupAccess);
+
     this.auditService.log({
       event: AuditEvent.GROUP_DELETED,
       resourceType: AuditResource.GROUP,

apps/server/src/core/page/page-access/page-access.service.ts

@@ -7,6 +7,7 @@ import {
   SpaceCaslSubject,
 } from '../../casl/interfaces/space-ability.type';
 import { SpaceRepo } from '@docmost/db/repos/space/space.repo';
+import { CacheOptions } from '../../../common/helpers/with-cache';
 
 @Injectable()
 export class PageAccessService {
@@ -21,9 +22,17 @@ export class PageAccessService {
    * If page has restrictions: page-level permission determines access.
    * If no restrictions: space-level permission determines access.
    */
-  async validateCanView(page: Page, user: User): Promise<void> {
+  async validateCanView(
+    page: Page,
+    user: User,
+    opts?: CacheOptions,
+  ): Promise<void> {
     // TODO: cache by pageId and userId.
-    const ability = await this.spaceAbility.createForUser(user, page.spaceId);
+    const ability = await this.spaceAbility.createForUser(
+      user,
+      page.spaceId,
+      opts,
+    );
 
     // User must be at least a space member
     if (ability.cannot(SpaceCaslAction.Read, SpaceCaslSubject.Page)) {
@@ -33,6 +42,7 @@ export class PageAccessService {
     const canAccess = await this.pagePermissionRepo.canUserAccessPage(
       user.id,
       page.id,
+      opts,
     );
     if (!canAccess) {
       throw new ForbiddenException();

apps/server/src/core/page/services/page.service.ts

@@ -57,6 +57,8 @@ import { WatcherService } from '../../watcher/watcher.service';
 import { sql } from 'kysely';
 import { TransclusionService } from '../transclusion/transclusion.service';
 import { LabelRepo } from '@docmost/db/repos/label/label.repo';
+import { WsService } from '../../../ws/ws.service';
+import { CollabAccessService } from '../../../collaboration/services/collab-access.service';
 
 @Injectable()
 export class PageService {
@@ -76,6 +78,8 @@ export class PageService {
     private readonly watcherService: WatcherService,
     private readonly transclusionService: TransclusionService,
     private readonly labelRepo: LabelRepo,
+    private readonly wsService: WsService,
+    private readonly collabAccessService: CollabAccessService,
   ) {}
 
   async findById(
@@ -399,7 +403,7 @@ export class PageService {
   }
 
   async movePageToSpace(rootPage: Page, spaceId: string, userId: string) {
-    return executeTx(this.db, async (trx) => {
+    const result = await executeTx(this.db, async (trx) => {
       await this.pageRepo.lockPageHierarchySpaces(
         [rootPage.spaceId, spaceId],
         trx,
@@ -536,6 +540,12 @@ export class PageService {
 
       return { childPageIds };
     });
+
+    await this.collabAccessService.recheckAccess({ pageIds: [rootPage.id] });
+    await this.wsService.invalidateSpaceRestrictionCache(spaceId);
+    await this.wsService.removeBaseSubscribersWithoutAccess([rootPage.id]);
+
+    return result;
   }
 
   async duplicatePage(
@@ -852,7 +862,7 @@ export class PageService {
       throw new BadRequestException('A page cannot be its own parent');
     }
 
-    await executeTx(this.db, async (trx) => {
+    const mayLoseAccess = await executeTx(this.db, async (trx) => {
       await this.pageRepo.lockPageHierarchySpaces(
         [movedPage.spaceId],
         trx,
@@ -904,7 +914,21 @@ export class PageService {
         dto.pageId,
         trx,
       );
+
+      // a new parent only changes access in a space with restricted pages
+      return (
+        parentPageId !== undefined &&
+        (await this.pagePermissionRepo.hasRestrictedPagesInSpace(
+          currentPage.spaceId,
+          { trx },
+        ))
+      );
     });
+
+    if (mayLoseAccess) {
+      await this.collabAccessService.recheckAccess({ pageIds: [dto.pageId] });
+      await this.wsService.removeBaseSubscribersWithoutAccess([dto.pageId]);
+    }
   }
 
   async getPageBreadCrumbs(childPageId: string) {

apps/server/src/core/session/session.service.spec.ts

@@ -0,0 +1,49 @@
+import { SessionService } from './session.service';
+
+function makeService(revoked: boolean) {
+  const userSessionRepo = {
+    revokeById: jest.fn(async () => revoked),
+    revokeAllExceptCurrent: jest.fn(async () => undefined),
+  };
+  const wsService = {
+    disconnectSession: jest.fn(),
+    disconnectOtherSessions: jest.fn(),
+  };
+  const service = new SessionService(
+    {} as any,
+    userSessionRepo as any,
+    {} as any,
+    {} as any,
+    wsService as any,
+  );
+  return { service, wsService };
+}
+
+describe('SessionService socket disconnects', () => {
+  it('disconnects the sockets of a revoked session', async () => {
+    const { service, wsService } = makeService(true);
+
+    await service.revokeSession('session_1', 'user_1', 'ws_1');
+
+    expect(wsService.disconnectSession).toHaveBeenCalledWith('session_1');
+  });
+
+  it('leaves sockets alone when nothing was revoked for this user', async () => {
+    const { service, wsService } = makeService(false);
+
+    await service.revokeSession('someone_elses_session', 'user_1', 'ws_1');
+
+    expect(wsService.disconnectSession).not.toHaveBeenCalled();
+  });
+
+  it('disconnects every other session of the user on revoke-all', async () => {
+    const { service, wsService } = makeService(true);
+
+    await service.revokeAllOtherSessions('session_1', 'user_1', 'ws_1');
+
+    expect(wsService.disconnectOtherSessions).toHaveBeenCalledWith(
+      'user_1',
+      'session_1',
+    );
+  });
+});

apps/server/src/core/session/session.service.ts

@@ -10,6 +10,7 @@ import {
   AUDIT_CONTEXT_KEY,
 } from '../../common/middlewares/audit-context.middleware';
 import * as Bowser from 'bowser';
+import { WsService } from '../../ws/ws.service';
 
 const MAX_SESSIONS_PER_USER = 25;
 const RETENTION_DAYS = 7;
@@ -23,6 +24,7 @@ export class SessionService {
     private readonly userSessionRepo: UserSessionRepo,
     private readonly environmentService: EnvironmentService,
     private readonly cls: ClsService,
+    private readonly wsService: WsService,
   ) {}
 
   @Interval('session-cleanup', 24 * 60 * 60 * 1000)
@@ -86,7 +88,14 @@ export class SessionService {
     userId: string,
     workspaceId: string,
   ): Promise<void> {
-    await this.userSessionRepo.revokeById(sessionId, userId, workspaceId);
+    const revoked = await this.userSessionRepo.revokeById(
+      sessionId,
+      userId,
+      workspaceId,
+    );
+    if (revoked) {
+      this.wsService.disconnectSession(sessionId);
+    }
   }
 
   async revokeAllOtherSessions(
@@ -99,6 +108,7 @@ export class SessionService {
       userId,
       workspaceId,
     );
+    this.wsService.disconnectOtherSessions(userId, currentSessionId);
   }
 
   private parseDeviceName(userAgent: string | null): string | null {

apps/server/src/core/space/services/space-member.service.ts

@@ -24,6 +24,8 @@ import {
   AUDIT_SERVICE,
   IAuditService,
 } from '../../../integrations/audit/audit.service';
+import { WsService } from '../../../ws/ws.service';
+import { CollabAccessService } from '../../../collaboration/services/collab-access.service';
 
 @Injectable()
 export class SpaceMemberService {
@@ -35,6 +37,8 @@ export class SpaceMemberService {
     private favoriteRepo: FavoriteRepo,
     @InjectKysely() private readonly db: KyselyDB,
     @Inject(AUDIT_SERVICE) private readonly auditService: IAuditService,
+    private readonly wsService: WsService,
+    private readonly collabAccessService: CollabAccessService,
   ) {}
 
   async addUserToSpace(
@@ -283,6 +287,11 @@ export class SpaceMemberService {
       return { space, spaceMember };
     });
 
+    await this.collabAccessService.recheckAccess({ userIds: affectedUserIds });
+    await this.wsService.removeUsersWithoutSpaceAccess(affectedUserIds, [
+      dto.spaceId,
+    ]);
+
     this.auditService.log({
       event: AuditEvent.SPACE_MEMBER_REMOVED,
       resourceType: AuditResource.SPACE_MEMBER,
@@ -337,7 +346,7 @@ export class SpaceMemberService {
       }
 
       if (spaceMember.role === dto.role) {
-        return { changed: false, space, spaceMember };
+        return { changed: false, space, spaceMember, demotedUserIds: [] };
       }
 
       if (spaceMember.role === SpaceRole.ADMIN) {
@@ -351,14 +360,25 @@ export class SpaceMemberService {
         trx,
       );
 
-      return { changed: true, space, spaceMember };
+      let demotedUserIds: string[] = [];
+      if (dto.role === SpaceRole.READER) {
+        demotedUserIds = spaceMember.userId
+          ? [spaceMember.userId]
+          : await this.groupUserRepo.getUserIdsByGroupId(spaceMember.groupId, {
+              trx,
+            });
+      }
+
+      return { changed: true, space, spaceMember, demotedUserIds };
     });
 
     if (!result.changed) {
       return;
     }
 
-    const { space, spaceMember } = result;
+    const { space, spaceMember, demotedUserIds } = result;
+
+    await this.collabAccessService.recheckAccess({ userIds: demotedUserIds });
 
     this.auditService.log({
       event: AuditEvent.SPACE_MEMBER_ROLE_CHANGED,

apps/server/src/core/workspace/services/workspace.service.ts

@@ -50,6 +50,8 @@ import {
   AUDIT_SERVICE,
   IAuditService,
 } from '../../../integrations/audit/audit.service';
+import { WsService } from '../../../ws/ws.service';
+import { CollabAccessService } from '../../../collaboration/services/collab-access.service';
 
 @Injectable()
 export class WorkspaceService {
@@ -75,6 +77,8 @@ export class WorkspaceService {
     @InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue,
     @Inject(AUDIT_SERVICE) private readonly auditService: IAuditService,
     private userSessionRepo: UserSessionRepo,
+    private wsService: WsService,
+    private readonly collabAccessService: CollabAccessService,
   ) {}
 
   async findById(workspaceId: string) {
@@ -908,6 +912,9 @@ export class WorkspaceService {
       return user;
     });
 
+    this.wsService.disconnectUser(userId);
+    await this.collabAccessService.recheckAccess({ userIds: [userId] });
+
     this.auditService.log({
       event: AuditEvent.USER_DEACTIVATED,
       resourceType: AuditResource.USER,
@@ -1032,6 +1039,9 @@ export class WorkspaceService {
       return user;
     });
 
+    this.wsService.disconnectUser(userId);
+    await this.collabAccessService.recheckAccess({ userIds: [userId] });
+
     this.auditService.log({
       event: AuditEvent.USER_DELETED,
       resourceType: AuditResource.USER,

apps/server/src/database/repos/group/group-user.repo.ts

@@ -155,8 +155,11 @@ export class GroupUserRepo {
     );
   }
 
-  async getUserIdsByGroupId(groupId: string): Promise<string[]> {
-    const rows = await this.db
+  async getUserIdsByGroupId(
+    groupId: string,
+    opts?: { trx?: KyselyTransaction },
+  ): Promise<string[]> {
+    const rows = await dbOrTx(this.db, opts?.trx)
       .selectFrom('groupUsers')
       .select('userId')
       .where('groupId', '=', groupId)

apps/server/src/database/repos/page/page-permission.repo.ts

@@ -19,7 +19,7 @@ import {
   executeWithCursorPagination,
 } from '@docmost/db/pagination/cursor-pagination';
 import { PagePermissionMember } from './types/page-permission.types';
-import { withCache } from '../../../common/helpers/with-cache';
+import { CacheOptions, withCache } from '../../../common/helpers/with-cache';
 import {
   CacheKey,
   PERMISSION_CACHE_TTL_MS,
@@ -104,6 +104,22 @@ export class PagePermissionRepo {
       .executeTakeFirst();
   }
 
+  async getPageIdsByGroupIds(
+    groupIds: string[],
+    opts?: { trx?: KyselyTransaction },
+  ): Promise<string[]> {
+    if (groupIds.length === 0) return [];
+    const rows = await dbOrTx(this.db, opts?.trx)
+      .selectFrom('pagePermissions')
+      .innerJoin('pageAccess', 'pageAccess.id', 'pagePermissions.pageAccessId')
+      .select('pageAccess.pageId')
+      .distinct()
+      .where('pagePermissions.groupId', 'in', groupIds)
+      .execute();
+
+    return rows.map((row) => row.pageId);
+  }
+
   async deletePagePermissionByUserId(
     pageAccessId: string,
     userId: string,
@@ -368,8 +384,12 @@ export class PagePermissionRepo {
   /**
    * Check if user can access a page by verifying they have permission on ALL restricted ancestors.
    */
-  async canUserAccessPage(userId: string, pageId: string): Promise<boolean> {
-    const { canAccess } = await this.canUserEditPage(userId, pageId);
+  async canUserAccessPage(
+    userId: string,
+    pageId: string,
+    opts?: CacheOptions,
+  ): Promise<boolean> {
+    const { canAccess } = await this.canUserEditPage(userId, pageId, opts);
     return canAccess;
   }
 
@@ -383,6 +403,7 @@ export class PagePermissionRepo {
   async canUserEditPage(
     userId: string,
     pageId: string,
+    opts?: CacheOptions,
   ): Promise<{
     hasAnyRestriction: boolean;
     canAccess: boolean;
@@ -431,6 +452,7 @@ export class PagePermissionRepo {
           canEdit: row.canAccess && (row.canEdit ?? false),
         };
       },
+      opts,
     );
   }
 
@@ -886,8 +908,11 @@ export class PagePermissionRepo {
    * Check if any page in a space has restrictions.
    * Used as a quick check to skip heavy permission filtering when no restrictions exist.
    */
-  async hasRestrictedPagesInSpace(spaceId: string): Promise<boolean> {
-    const result = await this.db
+  async hasRestrictedPagesInSpace(
+    spaceId: string,
+    opts?: { trx?: KyselyTransaction },
+  ): Promise<boolean> {
+    const result = await dbOrTx(this.db, opts?.trx)
       .selectNoFrom((eb) =>
         eb
           .exists(

apps/server/src/database/repos/session/user-session.repo.ts

@@ -61,15 +61,17 @@ export class UserSessionRepo {
     id: string,
     userId: string,
     workspaceId: string,
-  ): Promise<void> {
-    await this.db
+  ): Promise<boolean> {
+    const revoked = await this.db
       .updateTable('userSessions')
       .set({ revokedAt: new Date() })
       .where('id', '=', id)
       .where('userId', '=', userId)
       .where('workspaceId', '=', workspaceId)
       .where('revokedAt', 'is', null)
-      .execute();
+      .returning('id')
+      .executeTakeFirst();
+    return !!revoked;
   }
 
   async revokeAllExceptCurrent(

apps/server/src/database/repos/space/space-member.repo.ts

@@ -15,7 +15,7 @@ import { MemberInfo, UserSpaceRole } from './types';
 import { executeWithCursorPagination } from '@docmost/db/pagination/cursor-pagination';
 import { GroupRepo } from '@docmost/db/repos/group/group.repo';
 import { SpaceRepo } from '@docmost/db/repos/space/space.repo';
-import { withCache } from '../../../common/helpers/with-cache';
+import { CacheOptions, withCache } from '../../../common/helpers/with-cache';
 import {
   CacheKey,
   PERMISSION_CACHE_TTL_MS,
@@ -228,6 +228,7 @@ export class SpaceMemberRepo {
   async getUserSpaceRoles(
     userId: string,
     spaceId: string,
+    opts?: CacheOptions,
   ): Promise<UserSpaceRole[]> {
     return withCache(
       this.cacheManager,
@@ -258,6 +259,7 @@ export class SpaceMemberRepo {
         }
         return roles;
       },
+      opts,
     );
   }
 
@@ -285,11 +287,16 @@ export class SpaceMemberRepo {
     return new Set(rows.map((r) => r.userId));
   }
 
-  async getSpaceIdsByGroupId(groupId: string): Promise<string[]> {
-    const rows = await this.db
+  async getSpaceIdsByGroupIds(
+    groupIds: string[],
+    opts?: { trx?: KyselyTransaction },
+  ): Promise<string[]> {
+    if (groupIds.length === 0) return [];
+    const rows = await dbOrTx(this.db, opts?.trx)
       .selectFrom('spaceMembers')
       .select('spaceId')
-      .where('groupId', '=', groupId)
+      .distinct()
+      .where('groupId', 'in', groupIds)
       .execute();
 
     return rows.map((r) => r.spaceId);

apps/server/src/ee

@@ -1 +1 @@
-Subproject commit 35b9514c059b4d01e06fde09e6f7aaf3b48f7d87
+Subproject commit 1da2915bf81a43f7f5169445cc2e096dd15c3ef2

apps/server/src/ws/base-realtime.bridge.ts

@@ -46,4 +46,15 @@ export class BaseRealtimeBridge {
   async handleDisconnect(client: Socket): Promise<void> {
     await this.resolve()?.handleDisconnect(client);
   }
+
+  async removeUsersFromSpaceBases(
+    userIds: string[],
+    spaceId: string,
+  ): Promise<void> {
+    await this.resolve()?.removeUsersFromSpaceBases(userIds, spaceId);
+  }
+
+  async removeSubscribersWithoutAccess(pageIds: string[]): Promise<void> {
+    await this.resolve()?.removeSubscribersWithoutAccess(pageIds);
+  }
 }

apps/server/src/ws/ws.gateway.ts

@@ -10,10 +10,14 @@ import {
 import { Server, Socket } from 'socket.io';
 import { TokenService } from '../core/auth/services/token.service';
 import { JwtPayload, JwtType } from '../core/auth/dto/jwt-payload';
-import { OnModuleDestroy } from '@nestjs/common';
-import { SpaceMemberRepo } from '@docmost/db/repos/space/space-member.repo';
+import { OnModuleDestroy, UnauthorizedException } from '@nestjs/common';
+import { JsonWebTokenError } from '@nestjs/jwt';
 import { WsService } from './ws.service';
-import { getSpaceRoomName, getUserRoomName } from './ws.utils';
+import {
+  getSessionRoomName,
+  getSpaceRoomName,
+  getUserRoomName,
+} from './ws.utils';
 import { BaseRealtimeBridge } from './base-realtime.bridge';
 import * as cookie from 'cookie';
 
@@ -33,7 +37,6 @@ export class WsGateway
 
   constructor(
     private tokenService: TokenService,
-    private spaceMemberRepo: SpaceMemberRepo,
     private wsService: WsService,
     private baseRealtime: BaseRealtimeBridge,
   ) {}
@@ -45,7 +48,7 @@ export class WsGateway
 
   async handleConnection(client: Socket, ...args: any[]): Promise<void> {
     try {
-      const cookies = cookie.parse(client.handshake.headers.cookie);
+      const cookies = cookie.parse(client.handshake.headers.cookie ?? '');
       const token: JwtPayload = await this.tokenService.verifyJwt(
         cookies['authToken'],
         JwtType.ACCESS,
@@ -54,19 +57,30 @@ export class WsGateway
       const userId = token.sub;
       const workspaceId = token.workspaceId;
 
+      const userSpaceIds = await this.wsService.authorizeConnection(token);
+
       client.data.userId = userId;
       client.data.workspaceId = workspaceId;
 
-      const userSpaceIds = await this.spaceMemberRepo.getUserSpaceIds(userId);
-
       const userRoom = getUserRoomName(userId);
       const workspaceRoom = `workspace-${workspaceId}`;
       const spaceRooms = userSpaceIds.map((id) => getSpaceRoomName(id));
+      const sessionRooms = token.sessionId
+        ? [getSessionRoomName(token.sessionId)]
+        : [];
 
-      client.join([userRoom, workspaceRoom, ...spaceRooms]);
+      client.join([userRoom, workspaceRoom, ...spaceRooms, ...sessionRooms]);
     } catch (err) {
-      client.emit('Unauthorized');
-      client.disconnect();
+      if (
+        err instanceof UnauthorizedException ||
+        err instanceof JsonWebTokenError
+      ) {
+        client.emit('Unauthorized');
+        client.disconnect();
+        return;
+      }
+      // not an auth failure, closing the transport lets the client reconnect
+      client.conn.close();
     }
   }
 

apps/server/src/ws/ws.service.spec.ts

@@ -0,0 +1,67 @@
+import { WsService } from './ws.service';
+
+function makeService(accessBySpace: Record<string, string[]>) {
+  const spaceMemberRepo = {
+    getUserIdsWithSpaceAccess: jest.fn(
+      async (_userIds: string[], spaceId: string) =>
+        new Set(accessBySpace[spaceId] ?? []),
+    ),
+  };
+  const baseRealtime = {
+    removeUsersFromSpaceBases: jest.fn(async () => undefined),
+  };
+  const socketsLeave = jest.fn();
+  const server = { in: jest.fn(() => ({ socketsLeave })) };
+
+  const service = new WsService(
+    {} as any,
+    {} as any,
+    spaceMemberRepo as any,
+    baseRealtime as any,
+    {} as any,
+    {} as any,
+  );
+  service.setServer(server as any);
+
+  return { service, server, socketsLeave, baseRealtime };
+}
+
+describe('WsService.removeUsersWithoutSpaceAccess', () => {
+  it('removes users who lost access from the space room and its bases', async () => {
+    const { service, server, socketsLeave, baseRealtime } = makeService({
+      space_1: ['user_2'],
+    });
+
+    await service.removeUsersWithoutSpaceAccess(
+      ['user_1', 'user_2'],
+      ['space_1'],
+    );
+
+    expect(server.in).toHaveBeenCalledTimes(1);
+    expect(server.in).toHaveBeenCalledWith(['user-user_1']);
+    expect(socketsLeave).toHaveBeenCalledWith('space-space_1');
+    expect(baseRealtime.removeUsersFromSpaceBases).toHaveBeenCalledWith(
+      ['user_1'],
+      'space_1',
+    );
+  });
+
+  it('keeps users in spaces they can still access', async () => {
+    const { service, server, socketsLeave, baseRealtime } = makeService({
+      space_1: ['user_1'],
+    });
+
+    await service.removeUsersWithoutSpaceAccess(
+      ['user_1'],
+      ['space_1', 'space_2'],
+    );
+
+    expect(server.in).toHaveBeenCalledTimes(1);
+    expect(socketsLeave).toHaveBeenCalledWith('space-space_2');
+    expect(baseRealtime.removeUsersFromSpaceBases).toHaveBeenCalledTimes(1);
+    expect(baseRealtime.removeUsersFromSpaceBases).toHaveBeenCalledWith(
+      ['user_1'],
+      'space_2',
+    );
+  });
+});

apps/server/src/ws/ws.service.ts

@@ -1,32 +1,72 @@
-import { Inject, Injectable } from '@nestjs/common';
+import {
+  Inject,
+  Injectable,
+  Logger,
+  UnauthorizedException,
+} from '@nestjs/common';
 import { CACHE_MANAGER } from '@nestjs/cache-manager';
 import { Cache } from 'cache-manager';
 import { Server, Socket } from 'socket.io';
 import { PagePermissionRepo } from '@docmost/db/repos/page/page-permission.repo';
 import { SpaceMemberRepo } from '@docmost/db/repos/space/space-member.repo';
+import { UserRepo } from '@docmost/db/repos/user/user.repo';
+import { UserSessionRepo } from '@docmost/db/repos/session/user-session.repo';
+import { JwtPayload } from '../core/auth/dto/jwt-payload';
+import { isUserDisabled } from '../common/helpers';
 import { SpaceRole } from '../common/helpers/types/permission';
 import {
   TREE_EVENTS,
   WS_SPACE_RESTRICTION_CACHE_PREFIX,
   WS_CACHE_TTL_MS,
+  getSessionRoomName,
   getSpaceRoomName,
   getUserRoomName,
 } from './ws.utils';
+import { BaseRealtimeBridge } from './base-realtime.bridge';
 
 @Injectable()
 export class WsService {
+  private readonly logger = new Logger(WsService.name);
   private server: Server;
 
   constructor(
     private readonly pagePermissionRepo: PagePermissionRepo,
     @Inject(CACHE_MANAGER) private readonly cacheManager: Cache,
     private readonly spaceMemberRepo: SpaceMemberRepo,
+    private readonly baseRealtime: BaseRealtimeBridge,
+    private readonly userRepo: UserRepo,
+    private readonly userSessionRepo: UserSessionRepo,
   ) {}
 
   setServer(server: Server): void {
     this.server = server;
   }
 
+  async authorizeConnection(token: JwtPayload): Promise<string[]> {
+    const [user, session, spaceIds] = await Promise.all([
+      this.userRepo.findById(token.sub, token.workspaceId),
+      token.sessionId
+        ? this.userSessionRepo.findActiveById(token.sessionId)
+        : undefined,
+      this.spaceMemberRepo.getUserSpaceIds(token.sub),
+    ]);
+
+    if (!user || isUserDisabled(user)) {
+      throw new UnauthorizedException();
+    }
+
+    if (
+      token.sessionId &&
+      (!session ||
+        session.userId !== token.sub ||
+        session.workspaceId !== token.workspaceId)
+    ) {
+      throw new UnauthorizedException();
+    }
+
+    return spaceIds;
+  }
+
   async handleTreeEvent(client: Socket, data: any): Promise<void> {
     const room = getSpaceRoomName(data.spaceId);
 
@@ -73,9 +113,16 @@ export class WsService {
   }
 
   async invalidateSpaceRestrictionCache(spaceId: string): Promise<void> {
-    await this.cacheManager.del(
-      `${WS_SPACE_RESTRICTION_CACHE_PREFIX}${spaceId}`,
-    );
+    try {
+      await this.cacheManager.del(
+        `${WS_SPACE_RESTRICTION_CACHE_PREFIX}${spaceId}`,
+      );
+    } catch (err) {
+      this.logger.error(
+        { err },
+        `Failed to invalidate space restriction cache for ${spaceId}`,
+      );
+    }
   }
 
   async emitCommentEvent(
@@ -107,21 +154,72 @@ export class WsService {
     this.server.to(rooms).emit('message', data);
   }
 
+  disconnectUser(userId: string): void {
+    this.server.in(getUserRoomName(userId)).disconnectSockets(true);
+  }
+
+  disconnectSession(sessionId: string): void {
+    this.server.in(getSessionRoomName(sessionId)).disconnectSockets(true);
+  }
+
+  disconnectOtherSessions(userId: string, currentSessionId: string): void {
+    this.server
+      .in(getUserRoomName(userId))
+      .except(getSessionRoomName(currentSessionId))
+      .disconnectSockets(true);
+  }
+
+  async removeUsersWithoutSpaceAccess(
+    userIds: string[],
+    spaceIds: string[],
+  ): Promise<void> {
+    if (userIds.length === 0) return;
+
+    for (const spaceId of spaceIds) {
+      try {
+        const usersWithAccess =
+          await this.spaceMemberRepo.getUserIdsWithSpaceAccess(
+            userIds,
+            spaceId,
+          );
+        const revokedUserIds = userIds.filter(
+          (id) => !usersWithAccess.has(id),
+        );
+        if (revokedUserIds.length === 0) continue;
+
+        this.server
+          .in(revokedUserIds.map((id) => getUserRoomName(id)))
+          .socketsLeave(getSpaceRoomName(spaceId));
+        await this.baseRealtime.removeUsersFromSpaceBases(
+          revokedUserIds,
+          spaceId,
+        );
+      } catch (err) {
+        this.logger.error(
+          { err },
+          `Failed to remove revoked users from space ${spaceId} rooms`,
+        );
+      }
+    }
+  }
+
+  async removeBaseSubscribersWithoutAccess(pageIds: string[]): Promise<void> {
+    try {
+      await this.baseRealtime.removeSubscribersWithoutAccess(pageIds);
+    } catch (err) {
+      this.logger.error({ err }, 'Failed to remove base subscribers');
+    }
+  }
+
   async emitToSpaceExceptUsers(
     spaceId: string,
     excludeUserIds: string[],
     data: any,
   ): Promise<void> {
-    const room = getSpaceRoomName(spaceId);
-    const sockets = await this.server.in(room).fetchSockets();
-    const excludeSet = new Set(excludeUserIds);
-
-    for (const socket of sockets) {
-      const userId = socket.data.userId as string;
-      if (userId && !excludeSet.has(userId)) {
-        socket.emit('message', data);
-      }
-    }
+    this.server
+      .to(getSpaceRoomName(spaceId))
+      .except(excludeUserIds.map((id) => getUserRoomName(id)))
+      .emit('message', data);
   }
 
   isTreeEvent(data: any): boolean {

apps/server/src/ws/ws.utils.ts

@@ -9,6 +9,10 @@ export function getUserRoomName(userId: string): string {
   return `user-${userId}`;
 }
 
+export function getSessionRoomName(sessionId: string): string {
+  return `session-${sessionId}`;
+}
+
 export const TREE_EVENTS = new Set([
   'updateOne',
   'addTreeNode',