From 6fb6f6fce70a9898e1b9981108949c61df5daf24 Mon Sep 17 00:00:00 2001 From: "Claude Opus 4.8" Date: Thu, 9 Jul 2026 08:37:07 +0200 Subject: [PATCH] Add version snapshots: automatic, named, thinning (#41) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Version history is a core kickoff decision (ADR 0013). Snapshots are full, self-contained encoded Yjs states, so restore never depends on the update log and compaction (#40) cannot lose history. (The page_versions / page_pending_contributors tables and base schema landed a commit early, bundled into 3583a04; this commit completes #41.) - schema: page_versions gains created_by (editor of manual/pre-restore versions; null for automatic snapshots). shared: PageVersionView, CreateVersionInput, PageVersionTrigger. - collab: PostgresVersionStore tracks contributors per open doc (onChange), flushes them to the shared page_pending_contributors accumulator on store, creates an automatic snapshot on last-participant disconnect (only if something changed — no duplicate on a quick reconnect) and every 30 active-editing minutes. Contributors and snapshot are consumed atomically. - api: POST /pages/:id/versions creates a named version (write permission, label + creator, snapshot reconstructed from persisted state, consumes the same contributor accumulator). Daily version-thinning scheduler job keeps all versions for 90 days, then the newest auto snapshot per day; manual and pre-restore versions are never thinned. pre_restore trigger reserved for #42. Tests: collab (one auto version on session end with the full two-author contributor set, none when unchanged, no duplicate on reconnect, interval snapshot); api (named version stores label+creator, contributor set consumed, non-owner refused, thinning time-travel keeps newest-per-day beyond window). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01PGdhRiwU1WRL4XxJfZYipY --- .../migration.sql | 2 + apps/api/prisma/schema.prisma | 3 + apps/api/src/app.module.ts | 2 + apps/api/src/versions/versions.controller.ts | 22 +++ apps/api/src/versions/versions.module.ts | 32 ++++ .../src/versions/versions.service.db.test.ts | 170 ++++++++++++++++++ apps/api/src/versions/versions.service.ts | 142 +++++++++++++++ apps/collab/src/index.ts | 3 + apps/collab/src/server.ts | 25 ++- apps/collab/src/version-store.db.test.ts | 126 +++++++++++++ apps/collab/src/version-store.ts | 159 ++++++++++++++++ packages/shared/src/pages.ts | 22 +++ 12 files changed, 706 insertions(+), 2 deletions(-) create mode 100644 apps/api/prisma/migrations/20260709062727_page_version_created_by/migration.sql create mode 100644 apps/api/src/versions/versions.controller.ts create mode 100644 apps/api/src/versions/versions.module.ts create mode 100644 apps/api/src/versions/versions.service.db.test.ts create mode 100644 apps/api/src/versions/versions.service.ts create mode 100644 apps/collab/src/version-store.db.test.ts create mode 100644 apps/collab/src/version-store.ts diff --git a/apps/api/prisma/migrations/20260709062727_page_version_created_by/migration.sql b/apps/api/prisma/migrations/20260709062727_page_version_created_by/migration.sql new file mode 100644 index 0000000..633cba2 --- /dev/null +++ b/apps/api/prisma/migrations/20260709062727_page_version_created_by/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "page_versions" ADD COLUMN "created_by" TEXT; diff --git a/apps/api/prisma/schema.prisma b/apps/api/prisma/schema.prisma index f4a46d3..3ca1e14 100644 --- a/apps/api/prisma/schema.prisma +++ b/apps/api/prisma/schema.prisma @@ -148,6 +148,9 @@ model PageVersion { ydocSnapshot Bytes @map("ydoc_snapshot") trigger PageVersionTrigger label String? + /// Who created this version: the editor for `MANUAL`/`PRE_RESTORE`, null for + /// automatic snapshots (which have a contributor set instead of one author). + createdBy String? @map("created_by") contributorIds String[] @map("contributor_ids") createdAt DateTime @default(now()) @map("created_at") diff --git a/apps/api/src/app.module.ts b/apps/api/src/app.module.ts index c2c3a58..75cee38 100644 --- a/apps/api/src/app.module.ts +++ b/apps/api/src/app.module.ts @@ -18,6 +18,7 @@ import { RateLimitModule } from './rate-limit/rate-limit.module'; import { SettingsModule } from './settings/settings.module'; import { TrashModule } from './trash/trash.module'; import { UsersModule } from './users/users.module'; +import { VersionsModule } from './versions/versions.module'; @Module({ imports: [ @@ -32,6 +33,7 @@ import { UsersModule } from './users/users.module'; FilesModule, TrashModule, CompactionModule, + VersionsModule, AuthModule, AdminModule, LoggerModule.forRootAsync({ diff --git a/apps/api/src/versions/versions.controller.ts b/apps/api/src/versions/versions.controller.ts new file mode 100644 index 0000000..480304e --- /dev/null +++ b/apps/api/src/versions/versions.controller.ts @@ -0,0 +1,22 @@ +import { Body, Controller, Param, Post, Req } from '@nestjs/common'; +import { CreateVersionInput, PageVersionView, createVersionInputSchema } from '@dorfteich/shared'; + +import { AuthedRequest } from '../auth/auth.guard'; +import { ZodValidationPipe } from '../common/zod-validation.pipe'; +import { VersionsService } from './versions.service'; + +/** Page version history (issue #41, ADR 0013). List/get/restore arrive with #42. */ +@Controller() +export class VersionsController { + constructor(private readonly versions: VersionsService) {} + + /** Create a named version of the page. Requires write access. */ + @Post('pages/:id/versions') + async createNamed( + @Param('id') id: string, + @Body(new ZodValidationPipe(createVersionInputSchema)) input: CreateVersionInput, + @Req() request: AuthedRequest, + ): Promise { + return this.versions.createNamed(request.user!, id, input); + } +} diff --git a/apps/api/src/versions/versions.module.ts b/apps/api/src/versions/versions.module.ts new file mode 100644 index 0000000..f97f132 --- /dev/null +++ b/apps/api/src/versions/versions.module.ts @@ -0,0 +1,32 @@ +import { Module, OnModuleInit } from '@nestjs/common'; + +import { PondsModule } from '../ponds/ponds.module'; +import { SchedulerModule } from '../scheduler/scheduler.module'; +import { SchedulerService } from '../scheduler/scheduler.service'; + +import { VersionsController } from './versions.controller'; +import { VersionsService } from './versions.service'; + +/** Daily, per operations.md's maintenance-jobs table (ADR 0013 thinning). */ +const VERSION_THINNING_CADENCE_SECONDS = 24 * 60 * 60; + +@Module({ + imports: [PondsModule, SchedulerModule], + controllers: [VersionsController], + providers: [VersionsService], + exports: [VersionsService], +}) +export class VersionsModule implements OnModuleInit { + constructor( + private readonly scheduler: SchedulerService, + private readonly versions: VersionsService, + ) {} + + onModuleInit(): void { + this.scheduler.register({ + name: 'version-thinning', + cadenceSeconds: VERSION_THINNING_CADENCE_SECONDS, + run: () => this.versions.thinDueVersions().then(() => undefined), + }); + } +} diff --git a/apps/api/src/versions/versions.service.db.test.ts b/apps/api/src/versions/versions.service.db.test.ts new file mode 100644 index 0000000..fb00f79 --- /dev/null +++ b/apps/api/src/versions/versions.service.db.test.ts @@ -0,0 +1,170 @@ +import { randomUUID } from 'node:crypto'; + +import { INestApplication, NotFoundException } from '@nestjs/common'; +import { PrismaClient, User } from '@prisma/client'; +import { afterAll, 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 { VERSION_RETENTION_DAYS, VersionsService } from './versions.service'; + +describe.skipIf(!hasTestDb)('VersionsService (db, issue #41)', () => { + let app: INestApplication; + let prisma: PrismaClient; + let versions: VersionsService; + const suffix = uniqueSuffix(); + let owner: User; + let outsider: User; + let pondId: string; + const pageIds: string[] = []; + + async function createPage(): Promise { + const id = randomUUID(); + await prisma.page.create({ + data: { + id, + pondId, + title: 'Versioned', + slug: `p-${id.slice(0, 8)}`, + ydocState: new Uint8Array(Y.encodeStateAsUpdate(new Y.Doc())), + sortKey: 'a0', + createdBy: owner.id, + }, + }); + pageIds.push(id); + return id; + } + + beforeAll(async () => { + prisma = createTestPrisma(); + app = await createTestApp(); + versions = app.get(VersionsService); + + owner = await prisma.user.create({ + data: { + username: `ver-owner-${suffix}`, + email: `ver-owner-${suffix}@example.test`, + displayName: 'Version Owner', + }, + }); + outsider = await prisma.user.create({ + data: { + username: `ver-out-${suffix}`, + email: `ver-out-${suffix}@example.test`, + displayName: 'Version Outsider', + }, + }); + const pond = await prisma.pond.create({ + data: { + slug: `ver-pond-${suffix}`, + name: 'Version Pond', + type: 'PERSONAL', + ownerId: owner.id, + }, + }); + pondId = pond.id; + }); + + afterAll(async () => { + if (pageIds.length > 0) await prisma.page.deleteMany({ where: { id: { in: pageIds } } }); + await prisma.pond.deleteMany({ where: { id: pondId } }); + await prisma.user.deleteMany({ where: { id: { in: [owner.id, outsider.id] } } }); + await prisma.$disconnect(); + await app.close(); + }); + + it('creates a named version storing label and creator', async () => { + const pageId = await createPage(); + const view = await versions.createNamed(owner, pageId, { label: 'before restructuring' }); + + expect(view).toMatchObject({ + trigger: 'manual', + label: 'before restructuring', + createdBy: owner.id, + }); + const row = await prisma.pageVersion.findUniqueOrThrow({ where: { id: view.id } }); + expect(row.label).toBe('before restructuring'); + expect(row.createdBy).toBe(owner.id); + expect(row.trigger).toBe('MANUAL'); + expect(row.ydocSnapshot.byteLength).toBeGreaterThan(0); + }); + + it('captures and clears the pending contributor set', async () => { + const pageId = await createPage(); + const authorA = randomUUID(); + const authorB = randomUUID(); + await prisma.pagePendingContributor.createMany({ + data: [ + { pageId, userId: authorA }, + { pageId, userId: authorB }, + ], + }); + + const view = await versions.createNamed(owner, pageId, { label: 'snapshot' }); + expect([...view.contributorIds].sort()).toEqual([authorA, authorB].sort()); + // The accumulator is consumed so the next version does not re-attribute them. + expect(await prisma.pagePendingContributor.count({ where: { pageId } })).toBe(0); + }); + + it('refuses a named version for a user without write access', async () => { + const pageId = await createPage(); + await expect(versions.createNamed(outsider, pageId, { label: 'nope' })).rejects.toBeInstanceOf( + NotFoundException, + ); + expect(await prisma.pageVersion.count({ where: { pageId } })).toBe(0); + }); + + it('thins auto versions beyond the window to the newest per day, keeping the rest', async () => { + const pageId = await createPage(); + const daysAgo = (days: number, hour: number): Date => { + const d = new Date(); + d.setDate(d.getDate() - days); + d.setHours(hour, 0, 0, 0); + return d; + }; + const snapshot = new Uint8Array(Y.encodeStateAsUpdate(new Y.Doc())); + const auto = (createdAt: Date) => ({ + pageId, + ydocSnapshot: snapshot, + trigger: 'AUTO' as const, + contributorIds: [], + createdAt, + }); + + // Two auto versions on one day beyond the window (older + newer), one auto + // on another day beyond the window, one manual beyond the window, and one + // auto inside the window. + const oldDayOlder = daysAgo(120, 8); + const oldDayNewer = daysAgo(120, 20); + const otherOldDay = daysAgo(200, 12); + await prisma.pageVersion.createMany({ + data: [ + auto(oldDayOlder), + auto(oldDayNewer), + auto(otherOldDay), + { ...auto(daysAgo(150, 10)), trigger: 'MANUAL', label: 'keep me', createdBy: owner.id }, + auto(daysAgo(3, 10)), // within the window + ], + }); + + const removed = await versions.thinDueVersions(); + expect(removed).toBeGreaterThanOrEqual(1); + + const remaining = await prisma.pageVersion.findMany({ + where: { pageId }, + orderBy: { createdAt: 'asc' }, + select: { trigger: true, createdAt: true, label: true }, + }); + const times = remaining.map((v) => v.createdAt.getTime()); + // The older of the two same-day beyond-window auto versions is gone… + expect(times).not.toContain(oldDayOlder.getTime()); + // …its newer same-day sibling survives, as does the other-day one. + expect(times).toContain(oldDayNewer.getTime()); + expect(times).toContain(otherOldDay.getTime()); + // Manual and within-window auto are always kept. + expect(remaining.some((v) => v.label === 'keep me')).toBe(true); + expect(remaining.length).toBe(4); + expect(VERSION_RETENTION_DAYS).toBe(90); + }); +}); diff --git a/apps/api/src/versions/versions.service.ts b/apps/api/src/versions/versions.service.ts new file mode 100644 index 0000000..f603686 --- /dev/null +++ b/apps/api/src/versions/versions.service.ts @@ -0,0 +1,142 @@ +import { Injectable, NotFoundException } from '@nestjs/common'; +import { CreateVersionInput, PageVersionTrigger, PageVersionView } from '@dorfteich/shared'; +import { PageVersion, PageVersionTrigger as PrismaTrigger, User } from '@prisma/client'; +import { PinoLogger } from 'nestjs-pino'; +import * as Y from 'yjs'; + +import { InterimAccessService } from '../ponds/interim-access.service'; +import { PrismaService } from '../prisma/prisma.service'; + +/** + * Auto-versions older than this are thinned to one snapshot per day; everything + * within the window is kept in full (ADR 0013, default 90 days). Manual and + * pre-restore versions are intentional and never thinned. + */ +export const VERSION_RETENTION_DAYS = 90; + +const TRIGGER_TO_VIEW: Record = { + AUTO: 'auto', + MANUAL: 'manual', + PRE_RESTORE: 'pre_restore', +}; + +/** + * Page version history (issue #41, ADR 0013). Named versions are created here, + * permission-gated; automatic versions are created by the collab server. Both + * consume the shared `page_pending_contributors` accumulator so a version's + * contributor set reflects who edited since the previous version. The daily + * thinning job keeps history bounded. + */ +@Injectable() +export class VersionsService { + constructor( + private readonly prisma: PrismaService, + private readonly access: InterimAccessService, + private readonly logger: PinoLogger, + ) { + this.logger.setContext(VersionsService.name); + } + + viewOf(version: Omit): PageVersionView { + return { + id: version.id, + pageId: version.pageId, + trigger: TRIGGER_TO_VIEW[version.trigger], + label: version.label, + createdBy: version.createdBy, + contributorIds: version.contributorIds, + createdAt: version.createdAt.toISOString(), + }; + } + + /** + * Create a named version (requires write access, ADR 0013 / permissions.md). + * The snapshot is the page's current persisted state (base state plus the + * update log); it can lag the very latest live keystrokes by the collab + * store debounce, a deliberate simplification for a manual "save this state". + */ + async createNamed( + user: User, + pageId: string, + input: CreateVersionInput, + ): Promise { + const page = await this.prisma.page.findFirst({ + where: { id: pageId, deletedAt: null }, + include: { pond: true }, + }); + if (!page) throw new NotFoundException(); + this.access.assertCanModify(user, page.pond); + + const snapshot = await this.reconstructSnapshot(pageId); + + const created = await this.prisma.$transaction(async (tx) => { + // Consume the contributors accumulated since the previous version. + const pending = await tx.pagePendingContributor.findMany({ where: { pageId } }); + const contributorIds = pending.map((row) => row.userId); + await tx.pagePendingContributor.deleteMany({ where: { pageId } }); + return tx.pageVersion.create({ + data: { + pageId, + ydocSnapshot: snapshot, + trigger: 'MANUAL', + label: input.label, + createdBy: user.id, + contributorIds, + }, + }); + }); + + this.logger.info( + { event: 'audit: version created', pageId, versionId: created.id, userId: user.id }, + 'named version created', + ); + return this.viewOf(created); + } + + /** Reconstruct the page's full current Yjs state (base + update log). */ + private async reconstructSnapshot(pageId: string): Promise> { + const page = await this.prisma.page.findUniqueOrThrow({ + where: { id: pageId }, + select: { ydocState: true }, + }); + const updates = await this.prisma.pageUpdate.findMany({ + where: { pageId }, + orderBy: { seq: 'asc' }, + select: { update: true }, + }); + const doc = new Y.Doc(); + try { + Y.applyUpdate(doc, new Uint8Array(page.ydocState)); + for (const row of updates) Y.applyUpdate(doc, new Uint8Array(row.update)); + // A fresh copy: Prisma's Bytes input type is Uint8Array, + // which the ArrayBufferLike-typed encode result does not satisfy. + return new Uint8Array(Y.encodeStateAsUpdate(doc)); + } finally { + doc.destroy(); + } + } + + /** + * Thin automatic versions older than the retention window down to the newest + * one per day, keeping every manual/pre-restore version and everything within + * the window (ADR 0013). Returns the number of versions removed. + */ + async thinDueVersions(): Promise { + const result = await this.prisma.$executeRaw` + DELETE FROM page_versions v + WHERE v.trigger = 'AUTO' + AND v.created_at < now() - make_interval(days => ${VERSION_RETENTION_DAYS}::int) + AND EXISTS ( + SELECT 1 FROM page_versions newer + WHERE newer.page_id = v.page_id + AND newer.trigger = 'AUTO' + AND newer.created_at < now() - make_interval(days => ${VERSION_RETENTION_DAYS}::int) + AND date_trunc('day', newer.created_at) = date_trunc('day', v.created_at) + AND newer.created_at > v.created_at + )`; + if (result > 0) { + this.logger.info({ event: 'version.thinning.run', removed: result }, 'thinned old versions'); + } + return result; + } +} diff --git a/apps/collab/src/index.ts b/apps/collab/src/index.ts index da1e6f2..f6938a8 100644 --- a/apps/collab/src/index.ts +++ b/apps/collab/src/index.ts @@ -7,6 +7,7 @@ import { createLogger } from './logger.js'; import { PostgresPagePersistence } from './persistence.js'; import { createCollabServer } from './server.js'; import { PostgresSessionRegistry } from './session-registry.js'; +import { PostgresVersionStore } from './version-store.js'; /** * Entry point of the collaboration server (ADR 0003). Validates the @@ -23,6 +24,7 @@ async function bootstrap(): Promise { // 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 versionStore = new PostgresVersionStore({ pool, logger }); const server = createCollabServer({ version: env.APP_VERSION, @@ -31,6 +33,7 @@ async function bootstrap(): Promise { pingDatabase: () => pingDatabase(pool), persistence: new PostgresPagePersistence(pool), sessionRegistry, + versionStore, }); // Terminate live sessions when access to a pond is revoked (issue #39). The diff --git a/apps/collab/src/server.ts b/apps/collab/src/server.ts index 430fb47..ae866f3 100644 --- a/apps/collab/src/server.ts +++ b/apps/collab/src/server.ts @@ -6,6 +6,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'; +import type { VersionStore } from './version-store.js'; /** Per-connection context returned by onAuthenticate and used by later hooks. */ export interface CollabContext { @@ -29,6 +30,11 @@ export interface CollabServerDeps { * which keeps the unit/integration server tests free of a DB dependency. */ sessionRegistry?: SessionRegistry; + /** + * Creates automatic version snapshots and tracks contributors (#41). + * Optional for the same reason as `sessionRegistry`. + */ + versionStore?: VersionStore; } /** @@ -49,7 +55,8 @@ export interface CollabErrorMessage { * cache (#35). */ export function createCollabServer(deps: CollabServerDeps): Server { - const { version, logger, tokenSecret, pingDatabase, persistence, sessionRegistry } = deps; + const { version, logger, tokenSecret, pingDatabase, persistence, sessionRegistry, versionStore } = + deps; return new Server({ name: 'dorfteich-collab', @@ -104,11 +111,20 @@ export function createCollabServer(deps: CollabServerDeps): Server { ); }, - async onDisconnect({ documentName, socketId, clientsCount }) { + /** Attribute each change to the editing user for version contributor sets (#41). */ + async onChange({ documentName, context }) { + if (context?.userId) versionStore?.recordContributor(documentName, context.userId); + }, + + async onDisconnect({ documentName, socketId, clientsCount, document }) { logger.info( { event: 'connection.close', documentName, socketId, clientsCount }, 'collaboration connection closed', ); + // The editing session ended: snapshot a version if anything changed (#41). + if (clientsCount === 0) { + await versionStore?.onSessionEnd(documentName, document); + } }, /** @@ -120,6 +136,8 @@ export function createCollabServer(deps: CollabServerDeps): Server { const loaded = await persistence.loadInto(documentName, document); // Advertise the open session so the compaction job skips this page (#40). sessionRegistry?.markOpen(documentName); + // Start the active-editing interval clock for automatic versions (#41). + versionStore?.noteOpened(documentName); logger.debug( { event: 'document.load', documentName, loaded }, loaded ? 'document loaded' : 'document not found, starting empty', @@ -165,6 +183,8 @@ export function createCollabServer(deps: CollabServerDeps): Server { }, 'document persisted', ); + // Flush contributors and take an interval snapshot if one is due (#41). + await versionStore?.onStore(documentName, document); }, /** Drop the per-document store bookkeeping once Hocuspocus unloads it. */ @@ -172,6 +192,7 @@ export function createCollabServer(deps: CollabServerDeps): Server { persistence.forget(documentName); // The session ended; let the compaction job consider this page again (#40). sessionRegistry?.markClosed(documentName); + versionStore?.forget(documentName); }, async onRequest({ request, response }) { diff --git a/apps/collab/src/version-store.db.test.ts b/apps/collab/src/version-store.db.test.ts new file mode 100644 index 0000000..c9151e6 --- /dev/null +++ b/apps/collab/src/version-store.db.test.ts @@ -0,0 +1,126 @@ +import { randomUUID } from 'node:crypto'; + +import { Pool } from 'pg'; +import { pino } from 'pino'; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'; +import * as Y from 'yjs'; + +import { PostgresVersionStore } from './version-store.js'; +import { collabTestDatabaseUrlOrUndefined } from './testing/test-db.js'; + +const url = collabTestDatabaseUrlOrUndefined; +const logger = pino({ enabled: false }); + +/** A small non-empty Yjs doc to snapshot. */ +function makeDoc(text: string): Y.Doc { + const doc = new Y.Doc(); + doc.getText('t').insert(0, text); + return doc; +} + +describe.skipIf(!url)('PostgresVersionStore (DB-backed, issue #41)', () => { + let pool: Pool; + const userId = randomUUID(); + const authorA = randomUUID(); + const authorB = randomUUID(); + const pondId = randomUUID(); + let pageId: string; + + beforeAll(async () => { + pool = new Pool({ connectionString: url }); + await pool.query( + 'INSERT INTO users (id, username, email, display_name) VALUES ($1, $2, $3, $4)', + [userId, `vs-${userId.slice(0, 8)}`, `${userId}@example.test`, 'Version Tester'], + ); + await pool.query( + `INSERT INTO ponds (id, slug, name, type, owner_id, updated_at) + VALUES ($1, $2, 'Version Pond', 'PERSONAL', $3, now())`, + [pondId, `vs-pond-${pondId.slice(0, 8)}`, userId], + ); + }); + + beforeEach(async () => { + // Fresh page per test so version/contributor state never leaks between them. + pageId = randomUUID(); + 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, + ], + ); + }); + + afterAll(async () => { + // Pages (one per test) cascade to versions + pending contributors. + await pool.query('DELETE FROM pages WHERE pond_id = $1', [pondId]); + await pool.query('DELETE FROM ponds WHERE id = $1', [pondId]); + await pool.query('DELETE FROM users WHERE id = $1', [userId]); + await pool.end(); + }); + + async function versionsOf( + id: string, + ): Promise<{ trigger: string; contributor_ids: string[]; label: string | null }[]> { + const res = await pool.query<{ + trigger: string; + contributor_ids: string[]; + label: string | null; + }>( + 'SELECT trigger, contributor_ids, label FROM page_versions WHERE page_id = $1 ORDER BY created_at', + [id], + ); + return res.rows; + } + + it('creates exactly one auto version on session end with the full contributor set', async () => { + const store = new PostgresVersionStore({ pool, logger }); + store.recordContributor(pageId, authorA); + store.recordContributor(pageId, authorB); + + const doc = makeDoc('two authors edited'); + await store.onSessionEnd(pageId, doc); + + const versions = await versionsOf(pageId); + expect(versions).toHaveLength(1); + expect(versions[0]!.trigger).toBe('AUTO'); + expect([...versions[0]!.contributor_ids].sort()).toEqual([authorA, authorB].sort()); + doc.destroy(); + }); + + it('creates no version when nothing changed (empty contributor set)', async () => { + const store = new PostgresVersionStore({ pool, logger }); + await store.onSessionEnd(pageId, makeDoc('untouched')); + expect(await versionsOf(pageId)).toHaveLength(0); + }); + + it('does not create a duplicate version on a quick reconnect with no edits', async () => { + const store = new PostgresVersionStore({ pool, logger }); + store.recordContributor(pageId, authorA); + await store.onSessionEnd(pageId, makeDoc('edited once')); + expect(await versionsOf(pageId)).toHaveLength(1); + + // Reconnect + disconnect without any edit: pending is empty, so no version. + store.noteOpened(pageId); + await store.onSessionEnd(pageId, makeDoc('edited once')); + expect(await versionsOf(pageId)).toHaveLength(1); + }); + + it('creates an interval snapshot from onStore when the interval has elapsed', async () => { + // intervalMs 0 => any store after an edit is due for an interval snapshot. + const store = new PostgresVersionStore({ pool, logger, intervalMs: 0 }); + store.noteOpened(pageId); + store.recordContributor(pageId, authorA); + + await store.onStore(pageId, makeDoc('interval edit')); + + const versions = await versionsOf(pageId); + expect(versions).toHaveLength(1); + expect(versions[0]!.trigger).toBe('AUTO'); + expect(versions[0]!.contributor_ids).toEqual([authorA]); + }); +}); diff --git a/apps/collab/src/version-store.ts b/apps/collab/src/version-store.ts new file mode 100644 index 0000000..a366f85 --- /dev/null +++ b/apps/collab/src/version-store.ts @@ -0,0 +1,159 @@ +import type { Pool } from 'pg'; +import type { Logger } from 'pino'; +import * as Y from 'yjs'; + +/** + * Automatic version snapshots created by the collab server (issue #41, ADR + * 0013). Contributors are tracked per open document: {@link recordContributor} + * collects the users who edit, {@link onStore} flushes them to the shared + * `page_pending_contributors` accumulator (so the api's named-version endpoint + * sees them too) and creates the periodic interval snapshot, and + * {@link onSessionEnd} creates the end-of-session snapshot. + * + * A snapshot is only written when there were edits since the previous version + * (the accumulator is non-empty), which both implements "skip no-op periods" + * and prevents duplicate versions on a quick reconnect with no changes. + */ +export interface VersionStore { + /** Attribute a change to a user (called from `onChange`). */ + recordContributor(pageId: string, userId: string): void; + /** Initialise interval bookkeeping when a document opens. */ + noteOpened(pageId: string): void; + /** Flush contributors and create an interval snapshot if one is due. */ + onStore(pageId: string, doc: Y.Doc): Promise; + /** Create the end-of-session snapshot when the last participant leaves. */ + onSessionEnd(pageId: string, doc: Y.Doc): Promise; + /** Drop per-document bookkeeping when the document unloads. */ + forget(pageId: string): void; +} + +export interface VersionStoreDeps { + pool: Pool; + logger: Logger; + /** Active-editing interval between automatic snapshots (default 30 min). */ + intervalMs?: number; +} + +const DEFAULT_INTERVAL_MS = 30 * 60 * 1000; + +export class PostgresVersionStore implements VersionStore { + private readonly intervalMs: number; + /** Users who have edited since the last flush, per open document. */ + private readonly dirty = new Map>(); + /** Timestamp (ms) of the last automatic snapshot, per open document. */ + private readonly lastVersionAt = new Map(); + + constructor(private readonly deps: VersionStoreDeps) { + this.intervalMs = deps.intervalMs ?? DEFAULT_INTERVAL_MS; + } + + recordContributor(pageId: string, userId: string): void { + let set = this.dirty.get(pageId); + if (!set) { + set = new Set(); + this.dirty.set(pageId, set); + } + set.add(userId); + } + + noteOpened(pageId: string): void { + this.lastVersionAt.set(pageId, Date.now()); + } + + async onStore(pageId: string, doc: Y.Doc): Promise { + await this.flushContributors(pageId); + const last = this.lastVersionAt.get(pageId) ?? Date.now(); + if (Date.now() - last >= this.intervalMs) { + const created = await this.createVersion(pageId, doc); + // Reset the interval clock only when a snapshot was actually written, so + // an idle-but-open page doesn't churn empty checks into versions. + if (created) this.lastVersionAt.set(pageId, Date.now()); + } + } + + async onSessionEnd(pageId: string, doc: Y.Doc): Promise { + await this.flushContributors(pageId); + await this.createVersion(pageId, doc); + } + + forget(pageId: string): void { + this.dirty.delete(pageId); + this.lastVersionAt.delete(pageId); + } + + /** Move the in-memory contributor set into the shared DB accumulator. */ + private async flushContributors(pageId: string): Promise { + const set = this.dirty.get(pageId); + if (!set || set.size === 0) return; + // Swap in a fresh set first so contributors arriving during the await are + // not lost with the ones being flushed. + this.dirty.set(pageId, new Set()); + const users = [...set]; + try { + await this.deps.pool.query( + `INSERT INTO page_pending_contributors (page_id, user_id) + SELECT $1, u FROM unnest($2::text[]) AS u + ON CONFLICT (page_id, user_id) DO NOTHING`, + [pageId, users], + ); + } catch (error) { + // Put the users back so the next flush retries them; never throw out of a + // hook (a page purged mid-session is the expected benign failure here). + const current = this.dirty.get(pageId) ?? new Set(); + for (const u of users) current.add(u); + this.dirty.set(pageId, current); + this.deps.logger.warn( + { event: 'version.contributors.flush_failed', pageId, err: (error as Error).message }, + 'could not flush contributors', + ); + } + } + + /** + * Write an automatic snapshot if there are pending contributors, consuming + * them atomically. Returns whether a version row was created. + */ + private async createVersion(pageId: string, doc: Y.Doc): Promise { + const snapshot = Buffer.from(Y.encodeStateAsUpdate(doc)); + const client = await this.deps.pool.connect(); + try { + await client.query('BEGIN'); + const pending = await client.query<{ user_id: string }>( + 'SELECT user_id FROM page_pending_contributors WHERE page_id = $1 FOR UPDATE', + [pageId], + ); + if (pending.rows.length === 0) { + await client.query('ROLLBACK'); + return false; // nothing changed since the last version + } + const contributors = pending.rows.map((row) => row.user_id); + // Guard the FK: skip (but still clear the accumulator) if the page was + // trashed/purged mid-session. + const inserted = await client.query( + `INSERT INTO page_versions (id, page_id, ydoc_snapshot, trigger, label, contributor_ids, created_at) + SELECT gen_random_uuid(), $1, $2, 'AUTO', NULL, $3::text[], now() + WHERE EXISTS (SELECT 1 FROM pages WHERE id = $1 AND deleted_at IS NULL)`, + [pageId, snapshot, contributors], + ); + await client.query('DELETE FROM page_pending_contributors WHERE page_id = $1', [pageId]); + await client.query('COMMIT'); + const created = (inserted.rowCount ?? 0) > 0; + if (created) { + this.deps.logger.debug( + { event: 'version.auto.created', pageId, contributors: contributors.length }, + 'automatic version snapshot created', + ); + } + return created; + } catch (error) { + await client.query('ROLLBACK').catch(() => undefined); + this.deps.logger.warn( + { event: 'version.auto.failed', pageId, err: (error as Error).message }, + 'could not create automatic version snapshot', + ); + return false; + } finally { + client.release(); + } + } +} diff --git a/packages/shared/src/pages.ts b/packages/shared/src/pages.ts index 89a39da..1f305aa 100644 --- a/packages/shared/src/pages.ts +++ b/packages/shared/src/pages.ts @@ -56,3 +56,25 @@ export interface PageView { export interface PageStateView extends PageView { state: string; } + +/** Why a version snapshot exists (ADR 0013): automatic (session end / active + * interval), a named manual snapshot, or the automatic pre-restore snapshot. */ +export type PageVersionTrigger = 'auto' | 'manual' | 'pre_restore'; + +export const createVersionInputSchema = z.object({ + label: z.string().trim().min(1, 'validation.required').max(100, 'validation.tooLong'), +}); +export type CreateVersionInput = z.infer; + +/** A version in the history list (issue #41; no snapshot bytes). */ +export interface PageVersionView { + id: string; + pageId: string; + trigger: PageVersionTrigger; + label: string | null; + /** Editor for manual/pre-restore versions; null for automatic snapshots. */ + createdBy: string | null; + /** Users who edited since the previous version. */ + contributorIds: string[]; + createdAt: string; +}