Add permission-revocation handling for live and offline sessions (#39)

Revoking write access must terminate live sessions and let a user with
pending offline edits export them rather than lose them silently.

Backend (generic, reused by M5 grants #53):
- packages/shared: POND_ACCESS_CHANGED_CHANNEL, the LISTEN/NOTIFY channel
  shared by api and collab.
- api: PondAccessNotifier emits pg_notify(pond_access_changed, pondId) on
  a permission-relevant change; the single generic seam for revocation.
  Wired into pond soft-delete as the interim trigger (see==modify until
  #53).
- collab: a dedicated-connection LISTEN listener (LISTEN is connection-
  bound, not pooled) that, on a notification, closes every open connection
  to the pond's open pages. Clients then reconnect and the api re-issues a
  token reflecting current access (downgrade to ro, or 403/404). Reconnects
  and re-LISTENs if its connection drops.

Frontend:
- use-collab-provider: a refused token (403/404) on (re)connect sets
  accessRevoked and stops the reconnect loop; exposes discardLocal.
- AccessRevokedDialog: keeps local content visible and offers Markdown
  copy/download (derived from the live editor doc, so offline edits are
  included) and an explicit discard that clears IndexedDB. de+en strings.

Tests: collab DB-backed integration test proves a direct NOTIFY closes a
live session within seconds (AC1) and leaves unrelated ponds untouched;
listener unit tests; api test asserts soft-delete fires the notifier;
web test for the export Markdown derivation.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PGdhRiwU1WRL4XxJfZYipY

apps/api/src/ponds/pond-access-notifier.service.ts

@@ -0,0 +1,43 @@
+import { POND_ACCESS_CHANGED_CHANNEL } from '@dorfteich/shared';
+import { Injectable } from '@nestjs/common';
+import { PinoLogger } from 'nestjs-pino';
+
+import { PrismaService } from '../prisma/prisma.service';
+
+/**
+ * Announces permission-relevant changes to a pond so live collaboration
+ * sessions can be re-validated (issue #39, permissions.md §Performance).
+ *
+ * It emits a PostgreSQL `NOTIFY` on {@link POND_ACCESS_CHANGED_CHANNEL} with
+ * the pond id as payload; the collab server listens on that channel and closes
+ * the open connections to the pond's pages, so every client reconnects and
+ * re-acquires a token reflecting the current access (downgrade to read-only or
+ * rejection).
+ *
+ * This is the single, generic seam for revocation. Under the interim access
+ * model (M2–M4) the only access-relevant change is a pond being trashed, so
+ * that is where {@link notifyAccessChanged} is called today. When the real
+ * role model lands (#53, M5) grant changes call the very same method — no
+ * change to the collab side is needed. Callers should invoke it *after* the
+ * access-changing write has committed.
+ */
+@Injectable()
+export class PondAccessNotifier {
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly logger: PinoLogger,
+  ) {
+    this.logger.setContext(PondAccessNotifier.name);
+  }
+
+  /**
+   * Notify listeners that the access situation of `pondId` changed. The
+   * channel name is a fixed identifier and cannot be parameterized in a
+   * `NOTIFY` statement, so `pg_notify(text, text)` is used with the pond id
+   * bound as a parameter — the payload never interpolates untrusted text.
+   */
+  async notifyAccessChanged(pondId: string): Promise<void> {
+    await this.prisma.$executeRaw`SELECT pg_notify(${POND_ACCESS_CHANGED_CHANNEL}, ${pondId})`;
+    this.logger.debug({ pondId }, 'notified pond access change');
+  }
+}

apps/api/src/ponds/ponds.e2e.db.test.ts

@@ -1,12 +1,13 @@
 import { INestApplication } from '@nestjs/common';
 import { PrismaClient } from '@prisma/client';
 import request from 'supertest';
-import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest';
 
 import { AuthTokensService } from '../auth/auth-tokens.service';
 import { createTestApp, sessionCookieOf } from '../testing/test-app';
 import { createTestPrisma, hasTestDb, uniqueSuffix } from '../testing/test-db';
 import { UsersService } from '../users/users.service';
+import { PondAccessNotifier } from './pond-access-notifier.service';
 
 describe.skipIf(!hasTestDb)('ponds (e2e, issue #21)', () => {
   let app: INestApplication;
@@ -227,6 +228,25 @@ describe.skipIf(!hasTestDb)('ponds (e2e, issue #21)', () => {
     expect(restored.body.deletedAt).toBeNull();
   });
 
+  it('notifies the collab server when a pond is trashed (issue #39)', async () => {
+    // The api emits the generic pond-access-change signal on soft-delete; the
+    // collab server listens and terminates the affected live sessions. Here we
+    // assert the api side fires it with the pond id; the NOTIFY→close delivery
+    // is proven end-to-end in the collab access-listener DB test.
+    const notifier = app.get(PondAccessNotifier);
+    const spy = vi.spyOn(notifier, 'notifyAccessChanged');
+    const created = await api()
+      .post('/api/v1/ponds')
+      .set('Cookie', ownerCookie)
+      .send({ name: `Signalteich ${suffix}` })
+      .expect(201);
+
+    await api().delete(`/api/v1/ponds/${created.body.id}`).set('Cookie', ownerCookie).expect(204);
+
+    expect(spy).toHaveBeenCalledWith(created.body.id);
+    spy.mockRestore();
+  });
+
   it('refuses to delete the personal pond', async () => {
     const list = await api().get('/api/v1/ponds').set('Cookie', ownerCookie).expect(200);
     const personal = list.body.find((p: { type: string }) => p.type === 'personal');

apps/api/src/ponds/ponds.module.ts

@@ -3,13 +3,14 @@ import { Module } from '@nestjs/common';
 import { QuotasModule } from '../quotas/quotas.module';
 
 import { InterimAccessService } from './interim-access.service';
+import { PondAccessNotifier } from './pond-access-notifier.service';
 import { PondsController } from './ponds.controller';
 import { PondsService } from './ponds.service';
 
 @Module({
   imports: [QuotasModule],
   controllers: [PondsController],
-  providers: [PondsService, InterimAccessService],
-  exports: [PondsService, InterimAccessService],
+  providers: [PondsService, InterimAccessService, PondAccessNotifier],
+  exports: [PondsService, InterimAccessService, PondAccessNotifier],
 })
 export class PondsModule {}

apps/api/src/ponds/ponds.service.ts

@@ -12,6 +12,7 @@ import { PinoLogger } from 'nestjs-pino';
 import { PrismaService } from '../prisma/prisma.service';
 import { QuotaService } from '../quotas/quota.service';
 import { InterimAccessService } from './interim-access.service';
+import { PondAccessNotifier } from './pond-access-notifier.service';
 
 @Injectable()
 export class PondsService {
@@ -19,6 +20,7 @@ export class PondsService {
     private readonly prisma: PrismaService,
     private readonly access: InterimAccessService,
     private readonly quotas: QuotaService,
+    private readonly accessNotifier: PondAccessNotifier,
     private readonly logger: PinoLogger,
   ) {
     this.logger.setContext(PondsService.name);
@@ -143,6 +145,10 @@ export class PondsService {
       data: { deletedAt: new Date(), deletedBy: user.id },
     });
     this.logger.info({ pondId: id, userId: user.id }, 'audit: pond trashed');
+    // Trashing a pond is the interim access-relevant change: revalidate any
+    // live collaboration sessions on its pages (issue #39). Real per-user
+    // grant revocation reuses this same notification from M5 on (#53).
+    await this.accessNotifier.notifyAccessChanged(id);
   }
 
   /** Site-Admin-only (guarded at the controller): the pond-level trash. */

apps/collab/src/access-listener.db.test.ts

@@ -0,0 +1,167 @@
+import { randomUUID } from 'node:crypto';
+
+import { HocuspocusProvider } from '@hocuspocus/provider';
+import { POND_ACCESS_CHANGED_CHANNEL } from '@dorfteich/shared';
+import { signCollabToken } from '@dorfteich/shared/token-crypto';
+import { Client, Pool } from 'pg';
+import { pino } from 'pino';
+import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+import * as Y from 'yjs';
+
+import { createAccessListener } from './access-listener.js';
+import { createCollabServer } from './server.js';
+import { freePort } from './testing/free-port.js';
+import { InMemoryPagePersistence } from './testing/fake-persistence.js';
+import { collabTestDatabaseUrlOrUndefined } from './testing/test-db.js';
+
+const url = collabTestDatabaseUrlOrUndefined;
+const secret = 'access-listener-test-secret-32ch!!';
+const logger = pino({ enabled: false });
+
+/** Poll `predicate` until it is true or the timeout elapses. */
+async function waitFor(predicate: () => boolean, timeoutMs = 5000): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (predicate()) return;
+    await new Promise((resolve) => setTimeout(resolve, 20));
+  }
+  throw new Error('timed out waiting for condition');
+}
+
+/**
+ * End-to-end proof that a pond access change terminates its live sessions
+ * (issue #39, acceptance criterion 1). A real Hocuspocus server with the access
+ * listener wired to PostgreSQL, a connected client, and a direct `NOTIFY` (the
+ * generic signal the api emits) — the client's connection must close within
+ * seconds so it reconnects and re-validates.
+ */
+describe.skipIf(!url)('access listener (DB-backed)', () => {
+  let pool: Pool;
+  let server: ReturnType<typeof createCollabServer>;
+  let listener: ReturnType<typeof createAccessListener>;
+  let wsUrl: string;
+  const userId = randomUUID();
+  const pondId = randomUUID();
+  const otherPondId = randomUUID();
+  const pageId = randomUUID();
+
+  beforeAll(async () => {
+    pool = new Pool({ connectionString: url });
+    await pool.query(
+      'INSERT INTO users (id, username, email, display_name) VALUES ($1, $2, $3, $4)',
+      [userId, `al-${userId.slice(0, 8)}`, `${userId}@example.test`, 'Access Listener Tester'],
+    );
+    for (const [id, slug] of [
+      [pondId, `al-pond-${pondId.slice(0, 8)}`],
+      [otherPondId, `al-pond-${otherPondId.slice(0, 8)}`],
+    ]) {
+      await pool.query(
+        `INSERT INTO ponds (id, slug, name, type, owner_id, updated_at)
+         VALUES ($1, $2, 'Access Listener Pond', 'PERSONAL', $3, now())`,
+        [id, slug, userId],
+      );
+    }
+    await pool.query(
+      `INSERT INTO pages (id, pond_id, title, slug, ydoc_state, sort_key, created_by, updated_at)
+       VALUES ($1, $2, 'Test', $3, $4, 'a0', $5, now())`,
+      [
+        pageId,
+        pondId,
+        `p-${pageId.slice(0, 8)}`,
+        Buffer.from(Y.encodeStateAsUpdate(new Y.Doc())),
+        userId,
+      ],
+    );
+
+    server = createCollabServer({
+      version: 'test',
+      logger,
+      tokenSecret: secret,
+      pingDatabase: async () => ({ ok: true }),
+      persistence: new InMemoryPagePersistence(),
+    });
+    listener = createAccessListener({
+      createClient: () => new Client({ connectionString: url }),
+      pool,
+      openDocumentNames: () => [...server.hocuspocus.documents.keys()],
+      closeConnections: (name) => server.hocuspocus.closeConnections(name),
+      logger,
+      reconnectDelayMs: 100,
+    });
+    const port = await freePort();
+    await server.listen(port);
+    await listener.start();
+    wsUrl = `ws://127.0.0.1:${port}`;
+  });
+
+  afterAll(async () => {
+    await listener.stop();
+    await server.destroy();
+    await pool.query('DELETE FROM pages WHERE id = $1', [pageId]);
+    await pool.query('DELETE FROM ponds WHERE id = ANY($1::text[])', [[pondId, otherPondId]]);
+    await pool.query('DELETE FROM users WHERE id = $1', [userId]);
+    await pool.end();
+  });
+
+  function connect(name: string): {
+    provider: HocuspocusProvider;
+    doc: Y.Doc;
+    synced: () => boolean;
+    closeCount: () => number;
+    destroy: () => void;
+  } {
+    const doc = new Y.Doc();
+    let synced = false;
+    let closeCount = 0;
+    const provider = new HocuspocusProvider({
+      url: wsUrl,
+      name,
+      document: doc,
+      token: signCollabToken({ userId, pageId: name, mode: 'rw' }, secret, 60),
+      onSynced: () => {
+        synced = true;
+      },
+      onClose: () => {
+        closeCount += 1;
+      },
+    });
+    return {
+      provider,
+      doc,
+      synced: () => synced,
+      closeCount: () => closeCount,
+      destroy: () => {
+        provider.destroy();
+        doc.destroy();
+      },
+    };
+  }
+
+  it('closes a live session within seconds when its pond access changes', async () => {
+    const client = connect(pageId);
+    await waitFor(client.synced);
+    await waitFor(() => server.hocuspocus.documents.has(pageId));
+    const closesBefore = client.closeCount();
+
+    // The generic signal the api emits on a permission-relevant change.
+    await pool.query(`SELECT pg_notify('${POND_ACCESS_CHANGED_CHANNEL}', $1)`, [pondId]);
+
+    await waitFor(() => client.closeCount() > closesBefore);
+    expect(client.closeCount()).toBeGreaterThan(closesBefore);
+    client.destroy();
+  });
+
+  it('leaves sessions of unrelated ponds untouched', async () => {
+    const client = connect(pageId);
+    await waitFor(client.synced);
+    await waitFor(() => server.hocuspocus.documents.has(pageId));
+    const closesBefore = client.closeCount();
+
+    await pool.query(`SELECT pg_notify('${POND_ACCESS_CHANGED_CHANNEL}', $1)`, [otherPondId]);
+    // Give the listener time to receive and (correctly) ignore the notification.
+    await new Promise((resolve) => setTimeout(resolve, 500));
+
+    expect(client.closeCount()).toBe(closesBefore);
+    client.destroy();
+  });
+});

apps/collab/src/access-listener.test.ts

@@ -0,0 +1,135 @@
+import { EventEmitter } from 'node:events';
+
+import { POND_ACCESS_CHANGED_CHANNEL } from '@dorfteich/shared';
+import type { Pool } from 'pg';
+import { pino } from 'pino';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+
+import { createAccessListener, type AccessListener } from './access-listener.js';
+
+const logger = pino({ enabled: false });
+
+/** Poll `predicate` until it is true or the timeout elapses. */
+async function waitFor(predicate: () => boolean, timeoutMs = 2000): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (predicate()) return;
+    await new Promise((resolve) => setTimeout(resolve, 5));
+  }
+  throw new Error('timed out waiting for condition');
+}
+
+/** A minimal stand-in for a `pg.Client` that lets tests drive notifications. */
+class FakeClient extends EventEmitter {
+  connect = vi.fn(async () => undefined);
+  query = vi.fn(async () => ({ rows: [] }));
+  end = vi.fn(async () => undefined);
+  /** Simulate a NOTIFY arriving on this connection. */
+  notify(channel: string, payload: string): void {
+    this.emit('notification', { channel, payload });
+  }
+}
+
+describe('access listener', () => {
+  const listeners: AccessListener[] = [];
+
+  afterEach(async () => {
+    await Promise.all(listeners.splice(0).map((listener) => listener.stop()));
+    vi.restoreAllMocks();
+  });
+
+  function build(overrides: { openDocumentNames?: () => string[]; poolRows?: { id: string }[] }) {
+    const client = new FakeClient();
+    const closeConnections = vi.fn();
+    const poolQuery = vi.fn(async () => ({ rows: overrides.poolRows ?? [] }));
+    const pool = { query: poolQuery } as unknown as Pool;
+    const listener = createAccessListener({
+      createClient: () => client as unknown as never,
+      pool,
+      openDocumentNames: overrides.openDocumentNames ?? (() => []),
+      closeConnections,
+      logger,
+      reconnectDelayMs: 10,
+    });
+    listeners.push(listener);
+    return { client, closeConnections, poolQuery, listener };
+  }
+
+  it('LISTENs on the shared channel once connected', async () => {
+    const { client, listener } = build({});
+    await listener.start();
+    expect(client.connect).toHaveBeenCalledOnce();
+    expect(client.query).toHaveBeenCalledWith(`LISTEN ${POND_ACCESS_CHANGED_CHANNEL}`);
+  });
+
+  it('closes connections for the pond pages that are currently open', async () => {
+    const { client, closeConnections, poolQuery } = (() =>
+      build({
+        openDocumentNames: () => ['page-a', 'page-b', 'page-c'],
+        poolRows: [{ id: 'page-a' }, { id: 'page-c' }],
+      }))();
+    await listeners.at(-1)!.start();
+
+    client.notify(POND_ACCESS_CHANGED_CHANNEL, 'pond-1');
+
+    await waitFor(() => closeConnections.mock.calls.length === 2);
+    expect(poolQuery).toHaveBeenCalledWith(expect.stringContaining('FROM pages'), [
+      'pond-1',
+      ['page-a', 'page-b', 'page-c'],
+    ]);
+    expect(closeConnections).toHaveBeenCalledWith('page-a');
+    expect(closeConnections).toHaveBeenCalledWith('page-c');
+    expect(closeConnections).not.toHaveBeenCalledWith('page-b');
+  });
+
+  it('does nothing when no documents are open (no db round-trip)', async () => {
+    const { client, closeConnections, poolQuery } = build({ openDocumentNames: () => [] });
+    await listeners.at(-1)!.start();
+
+    client.notify(POND_ACCESS_CHANGED_CHANNEL, 'pond-1');
+    // Give the async handler a chance to run before asserting it did nothing.
+    await new Promise((resolve) => setTimeout(resolve, 30));
+
+    expect(poolQuery).not.toHaveBeenCalled();
+    expect(closeConnections).not.toHaveBeenCalled();
+  });
+
+  it('ignores notifications on other channels and with an empty payload', async () => {
+    const { client, poolQuery } = build({ openDocumentNames: () => ['page-a'] });
+    await listeners.at(-1)!.start();
+
+    client.notify('some_other_channel', 'pond-1');
+    client.notify(POND_ACCESS_CHANGED_CHANNEL, '');
+    await new Promise((resolve) => setTimeout(resolve, 30));
+
+    expect(poolQuery).not.toHaveBeenCalled();
+  });
+
+  it('reconnects and re-LISTENs after the connection errors', async () => {
+    const clients: FakeClient[] = [];
+    const closeConnections = vi.fn();
+    const pool = { query: vi.fn(async () => ({ rows: [] })) } as unknown as Pool;
+    const listener = createAccessListener({
+      createClient: () => {
+        const client = new FakeClient();
+        clients.push(client);
+        return client as unknown as never;
+      },
+      pool,
+      openDocumentNames: () => [],
+      closeConnections,
+      logger,
+      reconnectDelayMs: 10,
+    });
+    listeners.push(listener);
+    await listener.start();
+    expect(clients).toHaveLength(1);
+
+    // Simulate a dropped connection.
+    clients[0]!.emit('error', new Error('connection terminated'));
+
+    await waitFor(() => clients.length === 2);
+    expect(clients[1]!.connect).toHaveBeenCalledOnce();
+    expect(clients[1]!.query).toHaveBeenCalledWith(`LISTEN ${POND_ACCESS_CHANGED_CHANNEL}`);
+  });
+});

apps/collab/src/access-listener.ts

@@ -0,0 +1,140 @@
+import { POND_ACCESS_CHANGED_CHANNEL } from '@dorfteich/shared';
+import type { Client, Pool } from 'pg';
+import type { Logger } from 'pino';
+
+export interface AccessListenerDeps {
+  /**
+   * Creates the dedicated `LISTEN` connection. `LISTEN` is bound to a single
+   * connection and cannot be served from the pool, so the listener owns its own
+   * client and asks for a fresh one on every (re)connect.
+   */
+  createClient: () => Client;
+  /** Pool for the small "which open pages belong to this pond" lookup. */
+  pool: Pool;
+  /** Page ids of the documents the collab server currently holds open. */
+  openDocumentNames: () => string[];
+  /** Close every live connection to the given document (page id). */
+  closeConnections: (documentName: string) => void;
+  logger: Logger;
+  /**
+   * Delay before re-establishing a dropped `LISTEN` connection. A dropped
+   * connection silently stops delivering notifications, so it must reconnect
+   * and re-`LISTEN`. Overridable to keep tests fast.
+   */
+  reconnectDelayMs?: number;
+}
+
+export interface AccessListener {
+  /** Open the dedicated connection and start listening. */
+  start(): Promise<void>;
+  /** Stop listening and close the dedicated connection. Idempotent. */
+  stop(): Promise<void>;
+}
+
+const DEFAULT_RECONNECT_DELAY_MS = 1000;
+
+/**
+ * Listens for pond access changes and terminates the affected live sessions
+ * (issue #39, permissions.md §Performance).
+ *
+ * The api emits a `NOTIFY` on {@link POND_ACCESS_CHANGED_CHANNEL} carrying a
+ * pond id whenever access to that pond changes (interim: pond soft-delete; from
+ * M5: grant changes, #53). On each notification the listener closes every open
+ * connection to that pond's pages; Hocuspocus resets those sockets, the clients
+ * reconnect and re-acquire a token, and the api re-validates access — yielding a
+ * read-only downgrade or an outright rejection without the collab server ever
+ * needing to know the permission rules itself.
+ */
+export function createAccessListener(deps: AccessListenerDeps): AccessListener {
+  const reconnectDelayMs = deps.reconnectDelayMs ?? DEFAULT_RECONNECT_DELAY_MS;
+  let client: Client | null = null;
+  let stopped = false;
+  let reconnectTimer: NodeJS.Timeout | null = null;
+
+  async function revalidatePond(pondId: string): Promise<void> {
+    const open = deps.openDocumentNames();
+    if (open.length === 0) return;
+
+    // `pages.id` is a Prisma `String @id`, i.e. Postgres `text` — cast the
+    // parameter array to `text[]`, not `uuid[]`.
+    const result = await deps.pool.query<{ id: string }>(
+      'SELECT id FROM pages WHERE pond_id = $1 AND id = ANY($2::text[])',
+      [pondId, open],
+    );
+    for (const row of result.rows) {
+      deps.logger.info(
+        { event: 'access.revalidate', pondId, documentName: row.id },
+        'closing live connections after pond access change',
+      );
+      deps.closeConnections(row.id);
+    }
+  }
+
+  function scheduleReconnect(): void {
+    if (stopped || reconnectTimer) return;
+    reconnectTimer = setTimeout(() => {
+      reconnectTimer = null;
+      void connect();
+    }, reconnectDelayMs);
+    // Don't keep the process alive just for a pending reconnect attempt.
+    reconnectTimer.unref?.();
+  }
+
+  async function connect(): Promise<void> {
+    if (stopped) return;
+    const next = deps.createClient();
+    // A connection-level error (db restart, network drop) ends this client;
+    // reconnect so notifications keep flowing. Never let it crash the process.
+    next.on('error', (error) => {
+      deps.logger.warn(
+        { event: 'access.listen.error', err: error.message },
+        'access listener connection error; will reconnect',
+      );
+      if (client === next) client = null;
+      scheduleReconnect();
+    });
+    next.on('notification', (message) => {
+      if (message.channel !== POND_ACCESS_CHANGED_CHANNEL || !message.payload) return;
+      void revalidatePond(message.payload).catch((error: unknown) => {
+        deps.logger.error(
+          { event: 'access.revalidate.failed', err: (error as Error).message },
+          'failed to revalidate connections after pond access change',
+        );
+      });
+    });
+
+    try {
+      await next.connect();
+      await next.query(`LISTEN ${POND_ACCESS_CHANGED_CHANNEL}`);
+      client = next;
+      deps.logger.info(
+        { event: 'access.listen.ready', channel: POND_ACCESS_CHANGED_CHANNEL },
+        'listening for pond access changes',
+      );
+    } catch (error) {
+      deps.logger.warn(
+        { event: 'access.listen.connect_failed', err: (error as Error).message },
+        'could not start access listener; will retry',
+      );
+      await next.end().catch(() => undefined);
+      scheduleReconnect();
+    }
+  }
+
+  return {
+    async start(): Promise<void> {
+      stopped = false;
+      await connect();
+    },
+    async stop(): Promise<void> {
+      stopped = true;
+      if (reconnectTimer) {
+        clearTimeout(reconnectTimer);
+        reconnectTimer = null;
+      }
+      const current = client;
+      client = null;
+      if (current) await current.end().catch(() => undefined);
+    },
+  };
+}

apps/collab/src/index.ts

@@ -1,5 +1,7 @@
 import { collabEnvSchema, parseEnv } from '@dorfteich/shared';
+import { Client } from 'pg';
 
+import { createAccessListener } from './access-listener.js';
 import { createPool, pingDatabase } from './db.js';
 import { createLogger } from './logger.js';
 import { PostgresPagePersistence } from './persistence.js';
@@ -24,12 +26,26 @@ async function bootstrap(): Promise<void> {
     persistence: new PostgresPagePersistence(pool),
   });
 
+  // Terminate live sessions when access to a pond is revoked (issue #39). The
+  // listener owns a dedicated connection because `LISTEN` is connection-bound
+  // and cannot be served from the pool.
+  const accessListener = createAccessListener({
+    createClient: () => new Client({ connectionString: env.DATABASE_URL }),
+    pool,
+    openDocumentNames: () => [...server.hocuspocus.documents.keys()],
+    closeConnections: (documentName) => server.hocuspocus.closeConnections(documentName),
+    logger,
+  });
+
   await server.listen(env.PORT);
+  await accessListener.start();
   logger.info({ event: 'listen', port: env.PORT }, 'collaboration server listening');
 
   const shutdown = (signal: NodeJS.Signals): void => {
     logger.info({ event: 'shutdown', signal }, 'shutting down');
-    void Promise.allSettled([server.destroy(), pool.end()]).then(() => process.exit(0));
+    void Promise.allSettled([accessListener.stop(), server.destroy(), pool.end()]).then(() =>
+      process.exit(0),
+    );
   };
   process.on('SIGTERM', () => shutdown('SIGTERM'));
   process.on('SIGINT', () => shutdown('SIGINT'));

apps/web/src/editor/AccessRevokedDialog.tsx

@@ -0,0 +1,82 @@
+import type { Editor } from '@tiptap/react';
+import { useState } from 'react';
+import { useTranslation } from 'react-i18next';
+
+import { deriveMarkdownFromEditorJSON } from './derive-markdown';
+
+/**
+ * Shown when edit access to the open page was revoked while the user had
+ * unsynced changes (issue #39, realtime-collaboration.md §Offline). The local
+ * content stays visible behind the dialog; the user can export it as Markdown
+ * before losing it, and discards the device-local copy only on an explicit
+ * choice.
+ */
+export function AccessRevokedDialog({
+  editor,
+  slug,
+  onDiscard,
+}: {
+  editor: Editor;
+  slug: string;
+  onDiscard: () => Promise<void>;
+}): React.JSX.Element {
+  const { t } = useTranslation('editor');
+  const [copyStatus, setCopyStatus] = useState<'idle' | 'copied' | 'error'>('idle');
+
+  // Derive Markdown from the live editor doc, so it reflects the local edits
+  // the server never received (see derive-markdown.ts for the schema handling).
+  function currentMarkdown(): string {
+    return deriveMarkdownFromEditorJSON(editor.getJSON());
+  }
+
+  async function copyMarkdown(): Promise<void> {
+    try {
+      await navigator.clipboard.writeText(currentMarkdown());
+      setCopyStatus('copied');
+    } catch {
+      setCopyStatus('error');
+    }
+    setTimeout(() => setCopyStatus('idle'), 2000);
+  }
+
+  function downloadMarkdown(): void {
+    const blob = new Blob([currentMarkdown()], { type: 'text/markdown;charset=utf-8' });
+    const url = URL.createObjectURL(blob);
+    const anchor = document.createElement('a');
+    anchor.href = url;
+    anchor.download = `${slug}.md`;
+    document.body.append(anchor);
+    anchor.click();
+    anchor.remove();
+    URL.revokeObjectURL(url);
+  }
+
+  async function discard(): Promise<void> {
+    if (!window.confirm(t('accessRevoked.discardConfirm'))) return;
+    await onDiscard();
+  }
+
+  return (
+    <div
+      className="editor-banner editor-banner--error access-revoked"
+      role="alertdialog"
+      aria-label={t('accessRevoked.title')}
+    >
+      <p className="access-revoked__title">{t('accessRevoked.title')}</p>
+      <p className="access-revoked__description">{t('accessRevoked.description')}</p>
+      <div className="access-revoked__actions">
+        <button type="button" className="button" onClick={() => void copyMarkdown()}>
+          {copyStatus === 'idle' && t('accessRevoked.copy')}
+          {copyStatus === 'copied' && t('accessRevoked.copied')}
+          {copyStatus === 'error' && t('accessRevoked.copyFailed')}
+        </button>
+        <button type="button" className="button" onClick={downloadMarkdown}>
+          {t('accessRevoked.download')}
+        </button>
+        <button type="button" className="button" onClick={() => void discard()}>
+          {t('accessRevoked.discard')}
+        </button>
+      </div>
+    </div>
+  );
+}

apps/web/src/editor/derive-markdown.test.ts

@@ -0,0 +1,34 @@
+import { describe, expect, it } from 'vitest';
+
+import { deriveMarkdownFromEditorJSON } from './derive-markdown';
+
+describe('deriveMarkdownFromEditorJSON (issue #39 export)', () => {
+  it('converts a TipTap JSON document to Markdown', () => {
+    // Shaped like `editor.getJSON()` output, including a mark, so the export
+    // that a revoked user triggers reflects the local content faithfully.
+    const json = {
+      type: 'doc',
+      content: [
+        { type: 'heading', attrs: { level: 2 }, content: [{ type: 'text', text: 'Title' }] },
+        {
+          type: 'paragraph',
+          content: [
+            { type: 'text', text: 'Some ' },
+            { type: 'text', marks: [{ type: 'bold' }], text: 'bold' },
+            { type: 'text', text: ' text.' },
+          ],
+        },
+      ],
+    };
+
+    const markdown = deriveMarkdownFromEditorJSON(json);
+
+    expect(markdown).toContain('## Title');
+    expect(markdown).toContain('**bold**');
+  });
+
+  it('handles an empty document without throwing', () => {
+    const json = { type: 'doc', content: [{ type: 'paragraph' }] };
+    expect(() => deriveMarkdownFromEditorJSON(json)).not.toThrow();
+  });
+});

apps/web/src/editor/derive-markdown.ts

@@ -0,0 +1,16 @@
+import { docToMarkdown, editorSchema } from '@dorfteich/shared';
+import { Node as ProseMirrorNode } from '@tiptap/pm/model';
+
+/**
+ * Turn a TipTap editor JSON document into Markdown (issue #39 export, #30).
+ *
+ * TipTap builds its own `Schema` instance (see spec-utils.ts), so the JSON has
+ * to be rehydrated against the canonical `editorSchema` from packages/shared
+ * before `docToMarkdown` — wrapping foreign-schema nodes directly fails
+ * ProseMirror's identity-based content validation. Kept as a standalone helper
+ * so the export can be tested without mounting a live editor.
+ */
+export function deriveMarkdownFromEditorJSON(json: unknown): string {
+  const doc = ProseMirrorNode.fromJSON(editorSchema, json);
+  return docToMarkdown(doc);
+}

apps/web/src/editor/use-collab-provider.ts

@@ -1,10 +1,10 @@
 import type { CollabTokenResponse } from '@dorfteich/shared';
 import { HocuspocusProvider } from '@hocuspocus/provider';
-import { useEffect, useState } from 'react';
+import { useCallback, useEffect, useRef, useState } from 'react';
 import { IndexeddbPersistence } from 'y-indexeddb';
 import * as Y from 'yjs';
 
-import { apiGet } from '../lib/api';
+import { ApiError, apiGet } from '../lib/api';
 
 /** What the editor shows about the live connection (issue #36). */
 export type ConnectionStatus = 'connecting' | 'connected' | 'reconnecting' | 'offline';
@@ -18,6 +18,14 @@ export interface CollabState {
   tooLarge: boolean;
   /** True when there are edits held only on this device (offline, #38). */
   localOnly: boolean;
+  /**
+   * Set when the api refuses a collaboration token (403/404) on (re)connect,
+   * i.e. edit access was revoked (issue #39). The local content stays visible
+   * so the user can export it; live sync stops.
+   */
+  accessRevoked: boolean;
+  /** Discard the device-local copy of this page (clears IndexedDB, #39). */
+  discardLocal: () => Promise<void>;
 }
 
 /** IndexedDB database name for a page's local Yjs persistence (issue #38). */
@@ -50,6 +58,10 @@ export function useCollabProvider(ydoc: Y.Doc | null, pageId: string): CollabSta
   const [mode, setMode] = useState<'rw' | 'ro' | null>(null);
   const [tooLarge, setTooLarge] = useState(false);
   const [hasUnsynced, setHasUnsynced] = useState(false);
+  const [accessRevoked, setAccessRevoked] = useState(false);
+  // Held so `discardLocal` can clear the current page's IndexedDB store even
+  // after `accessRevoked` has stopped live sync (issue #39).
+  const localPersistenceRef = useRef<IndexeddbPersistence | null>(null);
 
   useEffect(() => {
     if (!ydoc) return;
@@ -62,15 +74,31 @@ export function useCollabProvider(ydoc: Y.Doc | null, pageId: string): CollabSta
     // edits survive a reload and offline work (ADR 0003, #38). The provider
     // and IndexedDB share the same Y.Doc and merge conflict-free.
     const localPersistence = new IndexeddbPersistence(localDbName(pageId), ydoc);
+    localPersistenceRef.current = localPersistence;
+    setAccessRevoked(false);
 
     const instance = new HocuspocusProvider({
       url: collabWsUrl(),
       name: pageId,
       document: ydoc,
       token: async () => {
+        try {
           const response = await apiGet<CollabTokenResponse>(`/pages/${pageId}/collab-token`);
-        if (!disposed) setMode(response.mode);
+          if (!disposed) {
+            setMode(response.mode);
+            setAccessRevoked(false);
+          }
           return response.token;
+        } catch (error) {
+          // A refused token (403/404) means edit access was revoked while we
+          // were connected or offline (#39). Surface it so the editor can offer
+          // an export; the effect below then stops the reconnect loop. Rethrow
+          // so this connection attempt aborts rather than using a stale token.
+          if (error instanceof ApiError && (error.status === 403 || error.status === 404)) {
+            if (!disposed) setAccessRevoked(true);
+          }
+          throw error;
+        }
       },
       onStatus: ({ status }) => {
         if (disposed) return;
@@ -112,12 +140,20 @@ export function useCollabProvider(ydoc: Y.Doc | null, pageId: string): CollabSta
       } else {
         void localPersistence.destroy();
       }
+      if (localPersistenceRef.current === localPersistence) localPersistenceRef.current = null;
       setProvider(null);
       setSynced(false);
       setHasUnsynced(false);
     };
   }, [ydoc, pageId]);
 
+  // Once access is revoked, stop the reconnect loop: every reconnect would only
+  // fetch another refused token and hammer the api (issue #39). The local copy
+  // stays intact for export until the user leaves or explicitly discards it.
+  useEffect(() => {
+    if (accessRevoked && provider) provider.disconnect();
+  }, [accessRevoked, provider]);
+
   useEffect(() => {
     const goOnline = (): void => setOnline(true);
     const goOffline = (): void => setOnline(false);
@@ -144,5 +180,9 @@ export function useCollabProvider(ydoc: Y.Doc | null, pageId: string): CollabSta
   // currently in sync (offline or reconnecting).
   const localOnly = hasUnsynced && status !== 'connected';
 
-  return { provider, status, mode, tooLarge, localOnly };
+  const discardLocal = useCallback(async (): Promise<void> => {
+    await localPersistenceRef.current?.clearData();
+  }, []);
+
+  return { provider, status, mode, tooLarge, localOnly, accessRevoked, discardLocal };
 }

apps/web/src/pages/PageEditorPage.tsx

@@ -9,6 +9,7 @@ import * as Y from 'yjs';
 
 import { useAuth } from '../auth/auth-context';
 import { FormError } from '../components/forms';
+import { AccessRevokedDialog } from '../editor/AccessRevokedDialog';
 import { collaborationCaretFor } from '../editor/collaboration-caret';
 import { documentExtensions } from '../editor/document-extensions';
 import { ImageUpload } from '../editor/image-upload';
@@ -30,9 +31,18 @@ interface ResolvedPage {
   title: string;
 }
 
-function PageEditor({ page, mode }: { page: ResolvedPage; mode: Mode }): React.JSX.Element {
+function PageEditor({
+  page,
+  mode,
+  pondSlug,
+}: {
+  page: ResolvedPage;
+  mode: Mode;
+  pondSlug: string;
+}): React.JSX.Element {
   const { t } = useTranslation('editor');
   const { user } = useAuth();
+  const navigate = useNavigate();
 
   // Created and destroyed within the same effect (not `useMemo` + a separate
   // cleanup effect): React StrictMode's dev-only mount→cleanup→remount would
@@ -53,7 +63,14 @@ function PageEditor({ page, mode }: { page: ResolvedPage; mode: Mode }): React.J
 
   const collab = useCollabProvider(ydoc, page.id);
   const readOnly = collab.mode === 'ro';
-  const canEdit = mode === 'edit' && !readOnly;
+  // A revoked page can no longer be edited (issue #39); the content stays
+  // visible for export via the dialog below.
+  const canEdit = mode === 'edit' && !readOnly && !collab.accessRevoked;
+
+  async function discardLocalAndLeave(): Promise<void> {
+    await collab.discardLocal();
+    navigate(`/p/${pondSlug}`);
+  }
 
   const editor = useEditor(
     {
@@ -106,6 +123,9 @@ function PageEditor({ page, mode }: { page: ResolvedPage; mode: Mode }): React.J
           {t('tooLarge.notice')}
         </div>
       )}
+      {collab.accessRevoked && (
+        <AccessRevokedDialog editor={editor} slug={page.slug} onDiscard={discardLocalAndLeave} />
+      )}
       <EditorContent editor={editor} className="editor-content" />
     </div>
   );
@@ -260,7 +280,7 @@ export function PageEditorPage(): React.JSX.Element {
         </button>
         <PageMenu pageId={resolved.id} slug={resolved.slug} pondSlug={pondSlug} />
       </div>
-      <PageEditor page={resolved} mode={mode} />
+      <PageEditor page={resolved} mode={mode} pondSlug={pondSlug} />
     </div>
   );
 }

apps/web/src/styles/base.css

@@ -642,6 +642,21 @@ button {
   color: var(--color-danger);
 }
 
+.access-revoked__title {
+  font-weight: 600;
+  margin: 0 0 var(--space-1);
+}
+
+.access-revoked__description {
+  margin: 0 0 var(--space-2);
+}
+
+.access-revoked__actions {
+  display: flex;
+  flex-wrap: wrap;
+  gap: var(--space-2);
+}
+
 .editor-content {
   padding: var(--space-4) var(--space-6);
   min-height: 12rem;

packages/shared/i18n/de/editor.json

@@ -27,6 +27,16 @@
   "offline": {
     "localOnly": "Diese Seite hat Änderungen, die nur auf diesem Gerät gespeichert sind. Sie werden automatisch synchronisiert, sobald du wieder online bist."
   },
+  "accessRevoked": {
+    "title": "Deine Bearbeitungsberechtigung wurde entfernt",
+    "description": "Du kannst diese Seite nicht mehr bearbeiten. Deine ungespeicherten Änderungen liegen noch auf diesem Gerät — exportiere sie, bevor du die Seite verlässt, da sie nicht auf dem Server gespeichert werden.",
+    "copy": "Als Markdown kopieren",
+    "copied": "Kopiert!",
+    "copyFailed": "Kopieren fehlgeschlagen",
+    "download": "Als Markdown herunterladen",
+    "discard": "Meine Änderungen verwerfen",
+    "discardConfirm": "Die auf diesem Gerät gespeicherten Änderungen verwerfen? Das kann nicht rückgängig gemacht werden."
+  },
   "toolbar": {
     "paragraph": "Absatz",
     "heading1": "Überschrift 1",

packages/shared/i18n/en/editor.json

@@ -27,6 +27,16 @@
   "offline": {
     "localOnly": "This page has changes saved only on this device. They'll sync automatically when you're back online."
   },
+  "accessRevoked": {
+    "title": "Your edit permission was removed",
+    "description": "You can no longer edit this page. Your unsaved changes are still on this device — export them before you leave, as they won't be saved to the server.",
+    "copy": "Copy as Markdown",
+    "copied": "Copied!",
+    "copyFailed": "Copy failed",
+    "download": "Download as Markdown",
+    "discard": "Discard my changes",
+    "discardConfirm": "Discard the changes saved on this device? This cannot be undone."
+  },
   "toolbar": {
     "paragraph": "Paragraph",
     "heading1": "Heading 1",

packages/shared/src/collab-token.ts

@@ -11,6 +11,17 @@ import { z } from 'zod';
 export const collabTokenModeSchema = z.enum(['rw', 'ro']);
 export type CollabTokenMode = z.infer<typeof collabTokenModeSchema>;
 
+/**
+ * PostgreSQL `LISTEN/NOTIFY` channel over which the api announces that the
+ * access situation of a pond changed (issue #39, permissions.md §Performance).
+ * The notification payload is the pond id. The api emits it whenever a
+ * permission-relevant change happens (interim: pond soft-delete; from M5 on:
+ * grant changes, #53); the collab server listens and re-validates every open
+ * connection to that pond's pages by closing them so clients reconnect and
+ * re-acquire a token reflecting the current access.
+ */
+export const POND_ACCESS_CHANGED_CHANNEL = 'pond_access_changed';
+
 /** The application claims carried by a collaboration token. */
 export const collabTokenClaimsSchema = z.object({
   userId: z.string().min(1),