Add Yjs update-log compaction job (#40)

Update logs grow with every edit; compaction bounds storage and load time.

- prisma: `collab_open_sessions` table (page_id, heartbeat_at) — the
  live-session registry that lets the compaction job avoid pages being
  edited, decoupled from collab (no api↔collab network call) and self-
  healing (a crashed collab's rows age out of the freshness window).
- collab: `PostgresSessionRegistry` marks a page open on document load and
  closed on unload, and refreshes an every-30s heartbeat for all open docs;
  wired into the server hooks and started/stopped in index.ts.
- api: `CompactionService` runs hourly via the shared scheduler (#31). For
  pages with > 500 log rows and no fresh session it merges `page_updates`
  into `ydoc_state` and deletes the merged rows in one FOR UPDATE
  transaction — atomic, so a mid-run crash leaves the page untouched and the
  next run resumes. Content is unchanged (merged state = base + all updates),
  so the content cache is left as-is; updated_at is deliberately not bumped.
  Metric log line with pages compacted / rows / bytes removed.

Tests: DB-backed compaction test (content hash unchanged, below-threshold
skipped, active session skipped then picked up next run, stale heartbeat
ignored, idempotent); session-registry DB test (open/close, heartbeat
refresh).

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

apps/api/prisma/migrations/20260709060328_collab_open_sessions/migration.sql

@@ -0,0 +1,10 @@
+-- CreateTable
+CREATE TABLE "collab_open_sessions" (
+    "page_id" TEXT NOT NULL,
+    "heartbeat_at" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "collab_open_sessions_pkey" PRIMARY KEY ("page_id")
+);
+
+-- CreateIndex
+CREATE INDEX "collab_open_sessions_heartbeat_at_idx" ON "collab_open_sessions"("heartbeat_at");

apps/api/prisma/schema.prisma

@@ -127,6 +127,20 @@ model PageUpdate {
   @@map("page_updates")
 }
 
+/// Live-session registry the collab server keeps current: one row per page
+/// with an open collaboration session, refreshed by a heartbeat (issue #40).
+/// The compaction job reads it to skip pages that are being edited; a stale
+/// row (collab crashed without unloading) ages out via the heartbeat window,
+/// so no FK to `pages` is needed and a leftover row is harmless. Collab is the
+/// only writer, over raw SQL (it does not use Prisma).
+model CollabOpenSession {
+  pageId      String   @id @map("page_id")
+  heartbeatAt DateTime @map("heartbeat_at")
+
+  @@index([heartbeatAt])
+  @@map("collab_open_sessions")
+}
+
 /// Derived plain representation refreshed on every state save (issue #23),
 /// built from the Yjs state via the shared editor schema. `outline` is the
 /// heading tree (`OutlineEntry[]` from @dorfteich/shared) as jsonb.

apps/api/src/app.module.ts

@@ -5,6 +5,7 @@ import { LoggerModule } from 'nestjs-pino';
 import { AdminModule } from './admin/admin.module';
 import { AuthModule } from './auth/auth.module';
 import { ApiExceptionFilter } from './common/api-exception.filter';
+import { CompactionModule } from './compaction/compaction.module';
 import { AppConfig } from './config/app-config.service';
 import { ConfigModule } from './config/config.module';
 import { FilesModule } from './files/files.module';
@@ -30,6 +31,7 @@ import { UsersModule } from './users/users.module';
     PagesModule,
     FilesModule,
     TrashModule,
+    CompactionModule,
     AuthModule,
     AdminModule,
     LoggerModule.forRootAsync({

apps/api/src/compaction/compaction.module.ts

@@ -0,0 +1,29 @@
+import { Module, OnModuleInit } from '@nestjs/common';
+
+import { SchedulerModule } from '../scheduler/scheduler.module';
+import { SchedulerService } from '../scheduler/scheduler.service';
+
+import { CompactionService } from './compaction.service';
+
+/** Hourly, per the issue #40 scope / operations.md maintenance-jobs table. */
+const COMPACTION_CADENCE_SECONDS = 60 * 60;
+
+@Module({
+  imports: [SchedulerModule],
+  providers: [CompactionService],
+  exports: [CompactionService],
+})
+export class CompactionModule implements OnModuleInit {
+  constructor(
+    private readonly scheduler: SchedulerService,
+    private readonly compaction: CompactionService,
+  ) {}
+
+  onModuleInit(): void {
+    this.scheduler.register({
+      name: 'page-compaction',
+      cadenceSeconds: COMPACTION_CADENCE_SECONDS,
+      run: () => this.compaction.compactDuePages().then(() => undefined),
+    });
+  }
+}

apps/api/src/compaction/compaction.service.db.test.ts

@@ -0,0 +1,174 @@
+import { randomUUID } from 'node:crypto';
+
+import { INestApplication } from '@nestjs/common';
+import { PrismaClient } from '@prisma/client';
+import { afterAll, afterEach, beforeAll, describe, expect, it } from 'vitest';
+import * as Y from 'yjs';
+
+import { createTestApp } from '../testing/test-app';
+import { createTestPrisma, hasTestDb, uniqueSuffix } from '../testing/test-db';
+import { COMPACTION_LOG_THRESHOLD, CompactionService } from './compaction.service';
+
+/** Reconstruct a page's document text from its stored state plus update log. */
+async function reconstructText(prisma: PrismaClient, pageId: string): Promise<string> {
+  const page = await prisma.page.findUniqueOrThrow({ where: { id: pageId } });
+  const updates = await prisma.pageUpdate.findMany({
+    where: { pageId },
+    orderBy: { seq: 'asc' },
+  });
+  const doc = new Y.Doc();
+  Y.applyUpdate(doc, new Uint8Array(page.ydocState));
+  for (const row of updates) Y.applyUpdate(doc, new Uint8Array(row.update));
+  const text = doc.getText('log').toString();
+  doc.destroy();
+  return text;
+}
+
+describe.skipIf(!hasTestDb)('CompactionService (db, issue #40)', () => {
+  let app: INestApplication;
+  let prisma: PrismaClient;
+  let compaction: CompactionService;
+  const suffix = uniqueSuffix();
+  let userId: string;
+  let pondId: string;
+  const pageIds: string[] = [];
+
+  beforeAll(async () => {
+    prisma = createTestPrisma();
+    app = await createTestApp();
+    compaction = app.get(CompactionService);
+
+    const user = await prisma.user.create({
+      data: {
+        username: `compaction-${suffix}`,
+        email: `compaction-${suffix}@example.test`,
+        displayName: 'Compaction Tester',
+      },
+    });
+    userId = user.id;
+    const pond = await prisma.pond.create({
+      data: {
+        slug: `compaction-pond-${suffix}`,
+        name: 'Compaction Pond',
+        type: 'PERSONAL',
+        ownerId: userId,
+      },
+    });
+    pondId = pond.id;
+  });
+
+  afterEach(async () => {
+    await prisma.collabOpenSession.deleteMany({ where: { pageId: { in: pageIds } } });
+    if (pageIds.length > 0) {
+      await prisma.page.deleteMany({ where: { id: { in: pageIds } } });
+      pageIds.length = 0;
+    }
+  });
+
+  afterAll(async () => {
+    await prisma.pond.deleteMany({ where: { id: pondId } });
+    await prisma.user.deleteMany({ where: { id: userId } });
+    await prisma.$disconnect();
+    await app.close();
+  });
+
+  /**
+   * Create a page whose update log has `logRows` entries, each a single
+   * character appended to a scratch text type — a well-formed incremental log
+   * over an empty base state, like a real editing session produces.
+   */
+  async function createPageWithLog(logRows: number): Promise<string> {
+    const id = randomUUID();
+    await prisma.page.create({
+      data: {
+        id,
+        pondId,
+        title: 'Test',
+        slug: `p-${id.slice(0, 8)}`,
+        ydocState: Buffer.from(Y.encodeStateAsUpdate(new Y.Doc())),
+        sortKey: 'a0',
+        createdBy: userId,
+      },
+    });
+    const master = new Y.Doc();
+    let vector = Y.encodeStateVector(master);
+    const rows: { id: string; pageId: string; seq: number; update: Buffer }[] = [];
+    for (let i = 0; i < logRows; i += 1) {
+      master.getText('log').insert(i, 'x');
+      rows.push({
+        id: randomUUID(),
+        pageId: id,
+        seq: i,
+        update: Buffer.from(Y.encodeStateAsUpdate(master, vector)),
+      });
+      vector = Y.encodeStateVector(master);
+    }
+    master.destroy();
+    await prisma.pageUpdate.createMany({ data: rows });
+    pageIds.push(id);
+    return id;
+  }
+
+  async function logCount(pageId: string): Promise<number> {
+    return prisma.pageUpdate.count({ where: { pageId } });
+  }
+
+  it('merges the log into ydoc_state without changing content (hash comparison)', async () => {
+    const pageId = await createPageWithLog(COMPACTION_LOG_THRESHOLD + 10);
+    const before = await reconstructText(prisma, pageId);
+
+    const summary = await compaction.compactDuePages();
+
+    expect(summary.pagesCompacted).toBeGreaterThanOrEqual(1);
+    expect(await logCount(pageId)).toBe(0);
+    // Content is identical after compaction (reconstructed from ydoc_state alone).
+    expect(await reconstructText(prisma, pageId)).toBe(before);
+    expect(before).toBe('x'.repeat(COMPACTION_LOG_THRESHOLD + 10));
+  });
+
+  it('leaves a page below the threshold untouched', async () => {
+    const pageId = await createPageWithLog(3);
+    await compaction.compactDuePages();
+    expect(await logCount(pageId)).toBe(3);
+  });
+
+  it('skips a page with an active session and compacts it on the next run', async () => {
+    const pageId = await createPageWithLog(COMPACTION_LOG_THRESHOLD + 5);
+    // A fresh heartbeat marks the page as being edited.
+    await prisma.collabOpenSession.create({
+      data: { pageId, heartbeatAt: new Date() },
+    });
+
+    await compaction.compactDuePages();
+    expect(await logCount(pageId)).toBe(COMPACTION_LOG_THRESHOLD + 5); // skipped
+
+    // Session ends: the row is removed (or ages out), and the next run compacts.
+    await prisma.collabOpenSession.deleteMany({ where: { pageId } });
+    await compaction.compactDuePages();
+    expect(await logCount(pageId)).toBe(0);
+  });
+
+  it('ignores a stale session heartbeat (crashed collab does not block forever)', async () => {
+    const pageId = await createPageWithLog(COMPACTION_LOG_THRESHOLD + 5);
+    // A row left behind by a crashed collab process: heartbeat far in the past.
+    await prisma.collabOpenSession.create({
+      data: { pageId, heartbeatAt: new Date(Date.now() - 60 * 60 * 1000) },
+    });
+
+    await compaction.compactDuePages();
+    expect(await logCount(pageId)).toBe(0);
+  });
+
+  it('is idempotent: a second run leaves the compacted page alone', async () => {
+    const pageId = await createPageWithLog(COMPACTION_LOG_THRESHOLD + 7);
+    const first = await compaction.compactDuePages();
+    expect(first.pagesCompacted).toBeGreaterThanOrEqual(1);
+    const afterFirst = await reconstructText(prisma, pageId);
+
+    // A second run must neither error nor touch the already-compacted page
+    // (the summary itself is DB-global, so we assert on this page specifically).
+    await compaction.compactDuePages();
+    expect(await logCount(pageId)).toBe(0);
+    expect(await reconstructText(prisma, pageId)).toBe(afterFirst);
+  });
+});

apps/api/src/compaction/compaction.service.ts

@@ -0,0 +1,150 @@
+import { Injectable } from '@nestjs/common';
+import { PinoLogger } from 'nestjs-pino';
+import * as Y from 'yjs';
+
+import { PrismaService } from '../prisma/prisma.service';
+
+/**
+ * When a page's append log grows past this many rows and no session is open,
+ * the log is merged back into `pages.ydoc_state` (ADR 0013 §compaction default
+ * 500). Collab merges inline at a lower threshold *within* a session
+ * (persistence.ts); this job is the cross-session bound for pages that
+ * accumulated a long log over many short sessions.
+ */
+export const COMPACTION_LOG_THRESHOLD = 500;
+
+/**
+ * A page counts as actively edited while its `collab_open_sessions` heartbeat
+ * is younger than this. It must comfortably exceed collab's heartbeat interval
+ * (30 s) so a still-open session is never mistaken for a crashed one; a crashed
+ * collab process's stale rows age past this window and stop blocking compaction.
+ */
+export const SESSION_FRESH_SECONDS = 90;
+
+export interface CompactionSummary {
+  pagesCompacted: number;
+  logRowsRemoved: number;
+  updateBytesRemoved: number;
+}
+
+/**
+ * Bounds Yjs update-log growth (issue #40, ADR 0013). Runs hourly via the
+ * shared scheduler (#31). For each eligible page it merges the `page_updates`
+ * log into `ydoc_state` and deletes the merged rows in one transaction, so a
+ * crash mid-run leaves the page untouched and the next run resumes it. The
+ * document content is unchanged — the merged state encodes exactly the base
+ * state plus every applied update — so the derived `page_content_cache` stays
+ * valid and is left alone.
+ */
+@Injectable()
+export class CompactionService {
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly logger: PinoLogger,
+  ) {
+    this.logger.setContext(CompactionService.name);
+  }
+
+  async compactDuePages(): Promise<CompactionSummary> {
+    const started = performance.now();
+    const candidates = await this.findCandidates();
+
+    const summary: CompactionSummary = {
+      pagesCompacted: 0,
+      logRowsRemoved: 0,
+      updateBytesRemoved: 0,
+    };
+    for (const pageId of candidates) {
+      const result = await this.compactPage(pageId);
+      if (result) {
+        summary.pagesCompacted += 1;
+        summary.logRowsRemoved += result.rowsRemoved;
+        summary.updateBytesRemoved += result.bytesRemoved;
+      }
+    }
+
+    if (summary.pagesCompacted > 0) {
+      this.logger.info(
+        {
+          event: 'compaction.run',
+          ...summary,
+          durationMs: Math.round(performance.now() - started),
+        },
+        'compacted page update logs',
+      );
+    }
+    return summary;
+  }
+
+  /**
+   * Pages whose log exceeds the threshold and that have no fresh collab
+   * session. The session check is repeated inside the per-page transaction;
+   * this scan only narrows the set cheaply.
+   */
+  private async findCandidates(): Promise<string[]> {
+    const rows = await this.prisma.$queryRaw<{ id: string }[]>`
+      SELECT p.id
+      FROM pages p
+      WHERE p.deleted_at IS NULL
+        AND (SELECT count(*) FROM page_updates u WHERE u.page_id = p.id) > ${COMPACTION_LOG_THRESHOLD}
+        AND NOT EXISTS (
+          SELECT 1 FROM collab_open_sessions s
+          WHERE s.page_id = p.id
+            AND s.heartbeat_at > now() - make_interval(secs => ${SESSION_FRESH_SECONDS})
+        )`;
+    return rows.map((row) => row.id);
+  }
+
+  /**
+   * Merge one page's log atomically. Returns null (a no-op) when the page
+   * vanished, gained a live session, or was already compacted by a concurrent
+   * run — which is what makes the job idempotent and crash-safe.
+   */
+  private async compactPage(
+    pageId: string,
+  ): Promise<{ rowsRemoved: number; bytesRemoved: number } | null> {
+    return this.prisma.$transaction(async (tx) => {
+      // Lock the page row for the whole merge: this serialises against collab's
+      // own store (which also locks the row before appending), so no update can
+      // be inserted or the page trashed underneath us.
+      const pageRows = await tx.$queryRaw<{ ydoc_state: Buffer }[]>`
+        SELECT ydoc_state FROM pages WHERE id = ${pageId} AND deleted_at IS NULL FOR UPDATE`;
+      const page = pageRows[0];
+      if (!page) return null;
+
+      // A session may have opened between the scan and acquiring the lock.
+      const active = await tx.$queryRaw<{ one: number }[]>`
+        SELECT 1 AS one FROM collab_open_sessions
+        WHERE page_id = ${pageId}
+          AND heartbeat_at > now() - make_interval(secs => ${SESSION_FRESH_SECONDS})
+        LIMIT 1`;
+      if (active.length > 0) return null;
+
+      const updates = await tx.$queryRaw<{ update: Buffer }[]>`
+        SELECT update FROM page_updates WHERE page_id = ${pageId} ORDER BY seq ASC`;
+      // Re-check under the lock: a concurrent run may have compacted already.
+      if (updates.length <= COMPACTION_LOG_THRESHOLD) return null;
+
+      const doc = new Y.Doc();
+      try {
+        Y.applyUpdate(doc, new Uint8Array(page.ydoc_state));
+        for (const row of updates) {
+          Y.applyUpdate(doc, new Uint8Array(row.update));
+        }
+        const merged = Buffer.from(Y.encodeStateAsUpdate(doc));
+
+        // Deliberately not touching updated_at: compaction is maintenance, not
+        // an edit, and must not look like a content change to sort/observers.
+        await tx.$executeRaw`UPDATE pages SET ydoc_state = ${merged} WHERE id = ${pageId}`;
+        // Safe to delete by page_id: the FOR UPDATE lock above means no new
+        // rows can have appeared since we read the log.
+        await tx.$executeRaw`DELETE FROM page_updates WHERE page_id = ${pageId}`;
+
+        const bytesRemoved = updates.reduce((sum, row) => sum + row.update.byteLength, 0);
+        return { rowsRemoved: updates.length, bytesRemoved };
+      } finally {
+        doc.destroy();
+      }
+    });
+  }
+}

apps/collab/src/index.ts

@@ -6,6 +6,7 @@ import { createPool, pingDatabase } from './db.js';
 import { createLogger } from './logger.js';
 import { PostgresPagePersistence } from './persistence.js';
 import { createCollabServer } from './server.js';
+import { PostgresSessionRegistry } from './session-registry.js';
 
 /**
  * Entry point of the collaboration server (ADR 0003). Validates the
@@ -18,12 +19,18 @@ async function bootstrap(): Promise<void> {
   const logger = createLogger(env);
   const pool = createPool(env.DATABASE_URL);
 
+  // Created before the server (the server's hooks call markOpen/markClosed);
+  // its heartbeat gets the server's open-document accessor at start() below,
+  // which sidesteps the mutual reference between the two.
+  const sessionRegistry = new PostgresSessionRegistry({ pool, logger });
+
   const server = createCollabServer({
     version: env.APP_VERSION,
     logger,
     tokenSecret: env.COLLAB_TOKEN_SECRET,
     pingDatabase: () => pingDatabase(pool),
     persistence: new PostgresPagePersistence(pool),
+    sessionRegistry,
   });
 
   // Terminate live sessions when access to a pond is revoked (issue #39). The
@@ -39,10 +46,12 @@ async function bootstrap(): Promise<void> {
 
   await server.listen(env.PORT);
   await accessListener.start();
+  sessionRegistry.start(() => [...server.hocuspocus.documents.keys()]);
   logger.info({ event: 'listen', port: env.PORT }, 'collaboration server listening');
 
   const shutdown = (signal: NodeJS.Signals): void => {
     logger.info({ event: 'shutdown', signal }, 'shutting down');
+    sessionRegistry.stop();
     void Promise.allSettled([accessListener.stop(), server.destroy(), pool.end()]).then(() =>
       process.exit(0),
     );

apps/collab/src/server.ts

@@ -5,6 +5,7 @@ import type { Logger } from 'pino';
 
 import { buildHealthReport, isHealthRequest, type DatabaseProbe } from './health.js';
 import type { PagePersistence } from './persistence.js';
+import type { SessionRegistry } from './session-registry.js';
 
 /** Per-connection context returned by onAuthenticate and used by later hooks. */
 export interface CollabContext {
@@ -22,6 +23,12 @@ export interface CollabServerDeps {
   pingDatabase: () => Promise<DatabaseProbe>;
   /** Loads and persists page documents against PostgreSQL (#35). */
   persistence: PagePersistence;
+  /**
+   * Advertises which pages have a live session so the compaction job skips
+   * them (#40). Optional: a server without it simply records no sessions,
+   * which keeps the unit/integration server tests free of a DB dependency.
+   */
+  sessionRegistry?: SessionRegistry;
 }
 
 /**
@@ -42,7 +49,7 @@ export interface CollabErrorMessage {
  * cache (#35).
  */
 export function createCollabServer(deps: CollabServerDeps): Server {
-  const { version, logger, tokenSecret, pingDatabase, persistence } = deps;
+  const { version, logger, tokenSecret, pingDatabase, persistence, sessionRegistry } = deps;
 
   return new Server({
     name: 'dorfteich-collab',
@@ -111,6 +118,8 @@ export function createCollabServer(deps: CollabServerDeps): Server {
      */
     async onLoadDocument({ documentName, document }) {
       const loaded = await persistence.loadInto(documentName, document);
+      // Advertise the open session so the compaction job skips this page (#40).
+      sessionRegistry?.markOpen(documentName);
       logger.debug(
         { event: 'document.load', documentName, loaded },
         loaded ? 'document loaded' : 'document not found, starting empty',
@@ -161,6 +170,8 @@ export function createCollabServer(deps: CollabServerDeps): Server {
     /** Drop the per-document store bookkeeping once Hocuspocus unloads it. */
     async afterUnloadDocument({ documentName }) {
       persistence.forget(documentName);
+      // The session ended; let the compaction job consider this page again (#40).
+      sessionRegistry?.markClosed(documentName);
     },
 
     async onRequest({ request, response }) {

apps/collab/src/session-registry.db.test.ts

@@ -0,0 +1,83 @@
+import { randomUUID } from 'node:crypto';
+
+import { Pool } from 'pg';
+import { pino } from 'pino';
+import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+
+import { PostgresSessionRegistry } from './session-registry.js';
+import { collabTestDatabaseUrlOrUndefined } from './testing/test-db.js';
+
+const url = collabTestDatabaseUrlOrUndefined;
+const logger = pino({ enabled: false });
+
+/** Poll `predicate` until it is true or the timeout elapses. */
+async function waitFor(predicate: () => Promise<boolean>, timeoutMs = 2000): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (await predicate()) return;
+    await new Promise((resolve) => setTimeout(resolve, 20));
+  }
+  throw new Error('timed out waiting for condition');
+}
+
+describe.skipIf(!url)('PostgresSessionRegistry (DB-backed, issue #40)', () => {
+  let pool: Pool;
+  const created: string[] = [];
+
+  beforeAll(() => {
+    pool = new Pool({ connectionString: url });
+  });
+
+  afterAll(async () => {
+    if (created.length > 0) {
+      await pool.query('DELETE FROM collab_open_sessions WHERE page_id = ANY($1::text[])', [
+        created,
+      ]);
+    }
+    await pool.end();
+  });
+
+  async function heartbeatOf(pageId: string): Promise<Date | null> {
+    const res = await pool.query<{ heartbeat_at: Date }>(
+      'SELECT heartbeat_at FROM collab_open_sessions WHERE page_id = $1',
+      [pageId],
+    );
+    return res.rows[0]?.heartbeat_at ?? null;
+  }
+
+  it('records an open session and clears it on close', async () => {
+    const pageId = randomUUID();
+    created.push(pageId);
+    const registry = new PostgresSessionRegistry({ pool, logger });
+
+    registry.markOpen(pageId);
+    await waitFor(async () => (await heartbeatOf(pageId)) !== null);
+    expect(await heartbeatOf(pageId)).toBeInstanceOf(Date);
+
+    registry.markClosed(pageId);
+    await waitFor(async () => (await heartbeatOf(pageId)) === null);
+    expect(await heartbeatOf(pageId)).toBeNull();
+  });
+
+  it('refreshes the heartbeat of open documents on its interval', async () => {
+    const pageId = randomUUID();
+    created.push(pageId);
+    const registry = new PostgresSessionRegistry({ pool, logger, heartbeatMs: 40 });
+
+    registry.markOpen(pageId);
+    await waitFor(async () => (await heartbeatOf(pageId)) !== null);
+    const first = await heartbeatOf(pageId);
+
+    registry.start(() => [pageId]);
+    try {
+      await waitFor(async () => {
+        const current = await heartbeatOf(pageId);
+        return current !== null && first !== null && current.getTime() > first.getTime();
+      });
+    } finally {
+      registry.stop();
+    }
+    const refreshed = await heartbeatOf(pageId);
+    expect(refreshed!.getTime()).toBeGreaterThan(first!.getTime());
+  });
+});

apps/collab/src/session-registry.ts

@@ -0,0 +1,116 @@
+import type { Pool } from 'pg';
+import type { Logger } from 'pino';
+
+/**
+ * The port the collab hooks use to advertise which pages have a live editing
+ * session (issue #40). The api's compaction job reads the backing table to
+ * skip pages that are being edited, so it never fights the collab writer.
+ */
+export interface SessionRegistry {
+  /** Record that a page now has an open session (called when its doc loads). */
+  markOpen(pageId: string): void;
+  /** Record that a page's session ended (called when its doc unloads). */
+  markClosed(pageId: string): void;
+}
+
+export interface SessionRegistryDeps {
+  pool: Pool;
+  logger: Logger;
+  /**
+   * How often the heartbeat refreshes the `heartbeat_at` of every open page,
+   * so the api can tell a still-open session from one whose collab process
+   * crashed without unloading (its row simply ages out). Overridable for tests.
+   */
+  heartbeatMs?: number;
+}
+
+export interface StartableSessionRegistry extends SessionRegistry {
+  /**
+   * Begin the periodic heartbeat. `openDocumentNames` returns the page ids the
+   * collab server currently holds open; taken here rather than in the
+   * constructor so the registry can be created before the server it reads from.
+   */
+  start(openDocumentNames: () => string[]): void;
+  /** Stop the heartbeat. Idempotent. */
+  stop(): void;
+}
+
+const DEFAULT_HEARTBEAT_MS = 30_000;
+
+/**
+ * PostgreSQL-backed {@link SessionRegistry} (issue #40, ADR 0013 §compaction).
+ *
+ * Each open page keeps a row in `collab_open_sessions` whose `heartbeat_at` is
+ * refreshed on a timer. The compaction job treats a page as actively edited
+ * while its heartbeat is fresh and ignores it otherwise, so a crashed collab
+ * process cannot block compaction forever — the stale rows just age out of the
+ * freshness window. Writes are best-effort: a failed heartbeat only risks a
+ * page being compacted a run later, never data loss.
+ */
+export class PostgresSessionRegistry implements StartableSessionRegistry {
+  private readonly heartbeatMs: number;
+  private timer: NodeJS.Timeout | undefined;
+  private openDocumentNames: () => string[] = () => [];
+
+  constructor(private readonly deps: SessionRegistryDeps) {
+    this.heartbeatMs = deps.heartbeatMs ?? DEFAULT_HEARTBEAT_MS;
+  }
+
+  markOpen(pageId: string): void {
+    void this.deps.pool
+      .query(
+        `INSERT INTO collab_open_sessions (page_id, heartbeat_at)
+         VALUES ($1, now())
+         ON CONFLICT (page_id) DO UPDATE SET heartbeat_at = now()`,
+        [pageId],
+      )
+      .catch((error: unknown) => {
+        this.deps.logger.warn(
+          { event: 'session.mark_open.failed', pageId, err: (error as Error).message },
+          'could not record open collab session',
+        );
+      });
+  }
+
+  markClosed(pageId: string): void {
+    void this.deps.pool
+      .query('DELETE FROM collab_open_sessions WHERE page_id = $1', [pageId])
+      .catch((error: unknown) => {
+        this.deps.logger.warn(
+          { event: 'session.mark_closed.failed', pageId, err: (error as Error).message },
+          'could not clear open collab session',
+        );
+      });
+  }
+
+  start(openDocumentNames: () => string[]): void {
+    this.openDocumentNames = openDocumentNames;
+    if (this.timer) return;
+    this.timer = setInterval(() => this.beat(), this.heartbeatMs);
+    // Don't keep the process alive just for the heartbeat.
+    this.timer.unref?.();
+  }
+
+  stop(): void {
+    if (this.timer) {
+      clearInterval(this.timer);
+      this.timer = undefined;
+    }
+  }
+
+  private beat(): void {
+    const open = this.openDocumentNames();
+    if (open.length === 0) return;
+    void this.deps.pool
+      .query(
+        `UPDATE collab_open_sessions SET heartbeat_at = now() WHERE page_id = ANY($1::text[])`,
+        [open],
+      )
+      .catch((error: unknown) => {
+        this.deps.logger.warn(
+          { event: 'session.heartbeat.failed', err: (error as Error).message },
+          'collab session heartbeat failed',
+        );
+      });
+  }
+}