Add version snapshots: automatic, named, thinning (#41)

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PGdhRiwU1WRL4XxJfZYipY

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;

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")
 

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({

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<PageVersionView> {
+    return this.versions.createNamed(request.user!, id, input);
+  }
+}

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),
+    });
+  }
+}

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<string> {
+    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);
+  });
+});

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<PrismaTrigger, PageVersionTrigger> = {
+  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<PageVersion, 'ydocSnapshot'>): 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<PageVersionView> {
+    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<Uint8Array<ArrayBuffer>> {
+    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<ArrayBuffer>,
+      // 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<number> {
+    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;
+  }
+}

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<void> {
   // 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<void> {
     pingDatabase: () => pingDatabase(pool),
     persistence: new PostgresPagePersistence(pool),
     sessionRegistry,
+    versionStore,
   });
 
   // Terminate live sessions when access to a pond is revoked (issue #39). The

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 }) {

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]);
+  });
+});

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<void>;
+  /** Create the end-of-session snapshot when the last participant leaves. */
+  onSessionEnd(pageId: string, doc: Y.Doc): Promise<void>;
+  /** 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<string, Set<string>>();
+  /** Timestamp (ms) of the last automatic snapshot, per open document. */
+  private readonly lastVersionAt = new Map<string, number>();
+
+  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<void> {
+    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<void> {
+    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<void> {
+    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<string>();
+      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<boolean> {
+    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();
+    }
+  }
+}

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<typeof createVersionInputSchema>;
+
+/** 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;
+}