From acd1cc32b7b5499be309338470270e30b00ed2c2 Mon Sep 17 00:00:00 2001 From: "Claude Opus 4.8" Date: Thu, 9 Jul 2026 08:15:55 +0200 Subject: [PATCH] Add Yjs update-log compaction job (#40) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01PGdhRiwU1WRL4XxJfZYipY --- .../migration.sql | 10 + apps/api/prisma/schema.prisma | 14 ++ apps/api/src/app.module.ts | 2 + apps/api/src/compaction/compaction.module.ts | 29 +++ .../compaction/compaction.service.db.test.ts | 174 ++++++++++++++++++ apps/api/src/compaction/compaction.service.ts | 150 +++++++++++++++ apps/collab/src/index.ts | 9 + apps/collab/src/server.ts | 13 +- apps/collab/src/session-registry.db.test.ts | 83 +++++++++ apps/collab/src/session-registry.ts | 116 ++++++++++++ 10 files changed, 599 insertions(+), 1 deletion(-) create mode 100644 apps/api/prisma/migrations/20260709060328_collab_open_sessions/migration.sql create mode 100644 apps/api/src/compaction/compaction.module.ts create mode 100644 apps/api/src/compaction/compaction.service.db.test.ts create mode 100644 apps/api/src/compaction/compaction.service.ts create mode 100644 apps/collab/src/session-registry.db.test.ts create mode 100644 apps/collab/src/session-registry.ts diff --git a/apps/api/prisma/migrations/20260709060328_collab_open_sessions/migration.sql b/apps/api/prisma/migrations/20260709060328_collab_open_sessions/migration.sql new file mode 100644 index 0000000..3df2e7c --- /dev/null +++ b/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"); diff --git a/apps/api/prisma/schema.prisma b/apps/api/prisma/schema.prisma index 72df17e..d0c8ac5 100644 --- a/apps/api/prisma/schema.prisma +++ b/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. diff --git a/apps/api/src/app.module.ts b/apps/api/src/app.module.ts index c063409..c2c3a58 100644 --- a/apps/api/src/app.module.ts +++ b/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({ diff --git a/apps/api/src/compaction/compaction.module.ts b/apps/api/src/compaction/compaction.module.ts new file mode 100644 index 0000000..3f46653 --- /dev/null +++ b/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), + }); + } +} diff --git a/apps/api/src/compaction/compaction.service.db.test.ts b/apps/api/src/compaction/compaction.service.db.test.ts new file mode 100644 index 0000000..fe234f8 --- /dev/null +++ b/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 { + 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 { + 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 { + 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); + }); +}); diff --git a/apps/api/src/compaction/compaction.service.ts b/apps/api/src/compaction/compaction.service.ts new file mode 100644 index 0000000..4aaea85 --- /dev/null +++ b/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 { + 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 { + 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(); + } + }); + } +} diff --git a/apps/collab/src/index.ts b/apps/collab/src/index.ts index e3a5e89..da1e6f2 100644 --- a/apps/collab/src/index.ts +++ b/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 { 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 { 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), ); diff --git a/apps/collab/src/server.ts b/apps/collab/src/server.ts index 2feb17c..430fb47 100644 --- a/apps/collab/src/server.ts +++ b/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; /** 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 }) { diff --git a/apps/collab/src/session-registry.db.test.ts b/apps/collab/src/session-registry.db.test.ts new file mode 100644 index 0000000..5096a1e --- /dev/null +++ b/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, timeoutMs = 2000): Promise { + 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 { + 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()); + }); +}); diff --git a/apps/collab/src/session-registry.ts b/apps/collab/src/session-registry.ts new file mode 100644 index 0000000..9a27a15 --- /dev/null +++ b/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', + ); + }); + } +}